aboutsummaryrefslogtreecommitdiffstats
path: root/vpp-api/java/m4/ax_prog_javadoc.m4
diff options
context:
space:
mode:
Diffstat (limited to 'vpp-api/java/m4/ax_prog_javadoc.m4')
0 files changed, 0 insertions, 0 deletions
124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 
# -*- coding: utf-8 -*-
#
# test/test_daemon.py
# Part of ‘python-daemon’, an implementation of PEP 3143.
#
# Copyright © 2008–2015 Ben Finney <ben+python@benfinney.id.au>
#
# This is free software: you may copy, modify, and/or distribute this work
# under the terms of the Apache License, version 2.0 as published by the
# Apache Software Foundation.
# No warranty expressed or implied. See the file ‘LICENSE.ASF-2’ for details.

""" Unit test for ‘daemon’ module.
    """

from __future__ import (absolute_import, unicode_literals)

import os
import sys
import tempfile
import resource
import errno
import signal
import socket
from types import ModuleType
import collections
import functools
try:
    # Standard library of Python 2.7 and later.
    from io import StringIO
except ImportError:
    # Standard library of Python 2.6 and earlier.
    from StringIO import StringIO

import mock

from . import scaffold
from .scaffold import (basestring, unicode)
from .test_pidfile import (
        FakeFileDescriptorStringIO,
        setup_pidfile_fixtures,
        )

import daemon


class ModuleExceptions_TestCase(scaffold.Exception_TestCase):
    """ Test cases for module exception classes. """

    scenarios = scaffold.make_exception_scenarios([
            ('daemon.daemon.DaemonError', dict(
                exc_type = daemon.daemon.DaemonError,
                min_args = 1,
                types = [Exception],
                )),
            ('daemon.daemon.DaemonOSEnvironmentError', dict(
                exc_type = daemon.daemon.DaemonOSEnvironmentError,
                min_args = 1,
                types = [daemon.daemon.DaemonError, OSError],
                )),
            ('daemon.daemon.DaemonProcessDetachError', dict(
                exc_type = daemon.daemon.DaemonProcessDetachError,
                min_args = 1,
                types = [daemon.daemon.DaemonError, OSError],
                )),
            ])


def setup_daemon_context_fixtures(testcase):
    """ Set up common test fixtures for DaemonContext test case.

        :param testcase: A ``TestCase`` instance to decorate.
        :return: ``None``.

        Decorate the `testcase` with fixtures for tests involving
        `DaemonContext`.

        """
    setup_streams_fixtures(testcase)

    setup_pidfile_fixtures(testcase)

    testcase.fake_pidfile_path = tempfile.mktemp()
    testcase.mock_pidlockfile = mock.MagicMock()
    testcase.mock_pidlockfile.path = testcase.fake_pidfile_path

    testcase.daemon_context_args = dict(
            stdin=testcase.stream_files_by_name['stdin'],
            stdout=testcase.stream_files_by_name['stdout'],
            stderr=testcase.stream_files_by_name['stderr'],
            )
    testcase.test_instance = daemon.DaemonContext(
            **testcase.daemon_context_args)

fake_default_signal_map = object()

@mock.patch.object(
        daemon.daemon, "is_detach_process_context_required",
        new=(lambda: True))
@mock.patch.object(
        daemon.daemon, "make_default_signal_map",
        new=(lambda: fake_default_signal_map))
@mock.patch.object(os, "setgid", new=(lambda x: object()))
@mock.patch.object(os, "setuid", new=(lambda x: object()))
class DaemonContext_BaseTestCase(scaffold.TestCase):
    """ Base class for DaemonContext test case classes. """

    def setUp(self):
        """ Set up test fixtures. """
        super(DaemonContext_BaseTestCase, self).setUp()

        setup_daemon_context_fixtures(self)


class DaemonContext_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext class. """

    def test_instantiate(self):
        """ New instance of DaemonContext should be created. """
        self.assertIsInstance(
                self.test_instance, daemon.daemon.DaemonContext)

    def test_minimum_zero_arguments(self):
        """ Initialiser should not require any arguments. """
        instance = daemon.daemon.DaemonContext()
        self.assertIsNot(instance, None)

    def test_has_specified_chroot_directory(self):
        """ Should have specified chroot_directory option. """
        args = dict(
                chroot_directory=object(),
                )
        expected_directory = args['chroot_directory']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_directory, instance.chroot_directory)

    def test_has_specified_working_directory(self):
        """ Should have specified working_directory option. """
        args = dict(
                working_directory=object(),
                )
        expected_directory = args['working_directory']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_directory, instance.working_directory)

    def test_has_default_working_directory(self):
        """ Should have default working_directory option. """
        args = dict()
        expected_directory = "/"
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_directory, instance.working_directory)

    def test_has_specified_creation_mask(self):
        """ Should have specified umask option. """
        args = dict(
                umask=object(),
                )
        expected_mask = args['umask']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_mask, instance.umask)

    def test_has_default_creation_mask(self):
        """ Should have default umask option. """
        args = dict()
        expected_mask = 0
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_mask, instance.umask)

    def test_has_specified_uid(self):
        """ Should have specified uid option. """
        args = dict(
                uid=object(),
                )
        expected_id = args['uid']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_id, instance.uid)

    def test_has_derived_uid(self):
        """ Should have uid option derived from process. """
        args = dict()
        expected_id = os.getuid()
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_id, instance.uid)

    def test_has_specified_gid(self):
        """ Should have specified gid option. """
        args = dict(
                gid=object(),
                )
        expected_id = args['gid']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_id, instance.gid)

    def test_has_derived_gid(self):
        """ Should have gid option derived from process. """
        args = dict()
        expected_id = os.getgid()
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_id, instance.gid)

    def test_has_specified_detach_process(self):
        """ Should have specified detach_process option. """
        args = dict(
                detach_process=object(),
                )
        expected_value = args['detach_process']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_value, instance.detach_process)

    def test_has_derived_detach_process(self):
        """ Should have detach_process option derived from environment. """
        args = dict()
        func = daemon.daemon.is_detach_process_context_required
        expected_value = func()
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_value, instance.detach_process)

    def test_has_specified_files_preserve(self):
        """ Should have specified files_preserve option. """
        args = dict(
                files_preserve=object(),
                )
        expected_files_preserve = args['files_preserve']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_files_preserve, instance.files_preserve)

    def test_has_specified_pidfile(self):
        """ Should have the specified pidfile. """
        args = dict(
                pidfile=object(),
                )
        expected_pidfile = args['pidfile']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_pidfile, instance.pidfile)

    def test_has_specified_stdin(self):
        """ Should have specified stdin option. """
        args = dict(
                stdin=object(),
                )
        expected_file = args['stdin']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_file, instance.stdin)

    def test_has_specified_stdout(self):
        """ Should have specified stdout option. """
        args = dict(
                stdout=object(),
                )
        expected_file = args['stdout']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_file, instance.stdout)

    def test_has_specified_stderr(self):
        """ Should have specified stderr option. """
        args = dict(
                stderr=object(),
                )
        expected_file = args['stderr']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_file, instance.stderr)

    def test_has_specified_signal_map(self):
        """ Should have specified signal_map option. """
        args = dict(
                signal_map=object(),
                )
        expected_signal_map = args['signal_map']
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_signal_map, instance.signal_map)

    def test_has_derived_signal_map(self):
        """ Should have signal_map option derived from system. """
        args = dict()
        expected_signal_map = daemon.daemon.make_default_signal_map()
        instance = daemon.daemon.DaemonContext(**args)
        self.assertEqual(expected_signal_map, instance.signal_map)


class DaemonContext_is_open_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext.is_open property. """

    def test_begin_false(self):
        """ Initial value of is_open should be False. """
        instance = self.test_instance
        self.assertEqual(False, instance.is_open)

    def test_write_fails(self):
        """ Writing to is_open should fail. """
        instance = self.test_instance
        self.assertRaises(
                AttributeError,
                setattr, instance, 'is_open', object())


class DaemonContext_open_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext.open method. """

    def setUp(self):
        """ Set up test fixtures. """
        super(DaemonContext_open_TestCase, self).setUp()

        self.test_instance._is_open = False

        self.mock_module_daemon = mock.MagicMock()
        daemon_func_patchers = dict(
                (func_name, mock.patch.object(
                    daemon.daemon, func_name))
                for func_name in [
                    "detach_process_context",
                    "change_working_directory",
                    "change_root_directory",
                    "change_file_creation_mask",
                    "change_process_owner",
                    "prevent_core_dump",
                    "close_all_open_files",
                    "redirect_stream",
                    "set_signal_handlers",
                    "register_atexit_function",
                    ])
        for (func_name, patcher) in daemon_func_patchers.items():
            mock_func = patcher.start()
            self.addCleanup(patcher.stop)
            self.mock_module_daemon.attach_mock(mock_func, func_name)

        self.mock_module_daemon.attach_mock(mock.Mock(), 'DaemonContext')

        self.test_files_preserve_fds = object()
        self.test_signal_handler_map = object()
        daemoncontext_method_return_values = {
                '_get_exclude_file_descriptors':
                    self.test_files_preserve_fds,
                '_make_signal_handler_map':
                    self.test_signal_handler_map,
                }
        daemoncontext_func_patchers = dict(
                (func_name, mock.patch.object(
                    daemon.daemon.DaemonContext,
                    func_name,
                    return_value=return_value))
                for (func_name, return_value) in
                    daemoncontext_method_return_values.items())
        for (func_name, patcher) in daemoncontext_func_patchers.items():
            mock_func = patcher.start()
            self.addCleanup(patcher.stop)
            self.mock_module_daemon.DaemonContext.attach_mock(
                    mock_func, func_name)

    def test_performs_steps_in_expected_sequence(self):
        """ Should perform daemonisation steps in expected sequence. """
        instance = self.test_instance
        instance.chroot_directory = object()
        instance.detach_process = True
        instance.pidfile = self.mock_pidlockfile
        self.mock_module_daemon.attach_mock(
                self.mock_pidlockfile, 'pidlockfile')
        expected_calls = [
                mock.call.change_root_directory(mock.ANY),
                mock.call.prevent_core_dump(),
                mock.call.change_file_creation_mask(mock.ANY),
                mock.call.change_working_directory(mock.ANY),
                mock.call.change_process_owner(mock.ANY, mock.ANY),
                mock.call.detach_process_context(),
                mock.call.DaemonContext._make_signal_handler_map(),
                mock.call.set_signal_handlers(mock.ANY),
                mock.call.DaemonContext._get_exclude_file_descriptors(),
                mock.call.close_all_open_files(exclude=mock.ANY),
                mock.call.redirect_stream(mock.ANY, mock.ANY),
                mock.call.redirect_stream(mock.ANY, mock.ANY),
                mock.call.redirect_stream(mock.ANY, mock.ANY),
                mock.call.pidlockfile.__enter__(),
                mock.call.register_atexit_function(mock.ANY),
                ]
        instance.open()
        self.mock_module_daemon.assert_has_calls(expected_calls)

    def test_returns_immediately_if_is_open(self):
        """ Should return immediately if is_open property is true. """
        instance = self.test_instance
        instance._is_open = True
        instance.open()
        self.assertEqual(0, len(self.mock_module_daemon.mock_calls))

    def test_changes_root_directory_to_chroot_directory(self):
        """ Should change root directory to `chroot_directory` option. """
        instance = self.test_instance
        chroot_directory = object()
        instance.chroot_directory = chroot_directory
        instance.open()
        self.mock_module_daemon.change_root_directory.assert_called_with(
                chroot_directory)

    def test_omits_chroot_if_no_chroot_directory(self):
        """ Should omit changing root directory if no `chroot_directory`. """
        instance = self.test_instance
        instance.chroot_directory = None
        instance.open()
        self.assertFalse(self.mock_module_daemon.change_root_directory.called)

    def test_prevents_core_dump(self):
        """ Should request prevention of core dumps. """
        instance = self.test_instance
        instance.open()
        self.mock_module_daemon.prevent_core_dump.assert_called_with()

    def test_omits_prevent_core_dump_if_prevent_core_false(self):
        """ Should omit preventing core dumps if `prevent_core` is false. """
        instance = self.test_instance
        instance.prevent_core = False
        instance.open()
        self.assertFalse(self.mock_module_daemon.prevent_core_dump.called)

    def test_closes_open_files(self):
        """ Should close all open files, excluding `files_preserve`. """
        instance = self.test_instance
        expected_exclude = self.test_files_preserve_fds
        instance.open()
        self.mock_module_daemon.close_all_open_files.assert_called_with(
                exclude=expected_exclude)

    def test_changes_directory_to_working_directory(self):
        """ Should change current directory to `working_directory` option. """
        instance = self.test_instance
        working_directory = object()
        instance.working_directory = working_directory
        instance.open()
        self.mock_module_daemon.change_working_directory.assert_called_with(
                working_directory)

    def test_changes_creation_mask_to_umask(self):
        """ Should change file creation mask to `umask` option. """
        instance = self.test_instance
        umask = object()
        instance.umask = umask
        instance.open()
        self.mock_module_daemon.change_file_creation_mask.assert_called_with(
                umask)

    def test_changes_owner_to_specified_uid_and_gid(self):
        """ Should change process UID and GID to `uid` and `gid` options. """
        instance = self.test_instance
        uid = object()
        gid = object()
        instance.uid = uid
        instance.gid = gid
        instance.open()
        self.mock_module_daemon.change_process_owner.assert_called_with(
                uid, gid)

    def test_detaches_process_context(self):
        """ Should request detach of process context. """
        instance = self.test_instance
        instance.open()
        self.mock_module_daemon.detach_process_context.assert_called_with()

    def test_omits_process_detach_if_not_required(self):
        """ Should omit detach of process context if not required. """
        instance = self.test_instance
        instance.detach_process = False
        instance.open()
        self.assertFalse(self.mock_module_daemon.detach_process_context.called)

    def test_sets_signal_handlers_from_signal_map(self):
        """ Should set signal handlers according to `signal_map`. """
        instance = self.test_instance
        instance.signal_map = object()
        expected_signal_handler_map = self.test_signal_handler_map
        instance.open()
        self.mock_module_daemon.set_signal_handlers.assert_called_with(
                expected_signal_handler_map)

    def test_redirects_standard_streams(self):
        """ Should request redirection of standard stream files. """
        instance = self.test_instance
        (system_stdin, system_stdout, system_stderr) = (
                sys.stdin, sys.stdout, sys.stderr)
        (target_stdin, target_stdout, target_stderr) = (
                self.stream_files_by_name[name]
                for name in ['stdin', 'stdout', 'stderr'])
        expected_calls = [
                mock.call(system_stdin, target_stdin),
                mock.call(system_stdout, target_stdout),
                mock.call(system_stderr, target_stderr),
                ]
        instance.open()
        self.mock_module_daemon.redirect_stream.assert_has_calls(
                expected_calls, any_order=True)

    def test_enters_pidfile_context(self):
        """ Should enter the PID file context manager. """
        instance = self.test_instance
        instance.pidfile = self.mock_pidlockfile
        instance.open()
        self.mock_pidlockfile.__enter__.assert_called_with()

    def test_sets_is_open_true(self):
        """ Should set the `is_open` property to True. """
        instance = self.test_instance
        instance.open()
        self.assertEqual(True, instance.is_open)

    def test_registers_close_method_for_atexit(self):
        """ Should register the `close` method for atexit processing. """
        instance = self.test_instance
        close_method = instance.close
        instance.open()
        self.mock_module_daemon.register_atexit_function.assert_called_with(
                close_method)


class DaemonContext_close_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext.close method. """

    def setUp(self):
        """ Set up test fixtures. """
        super(DaemonContext_close_TestCase, self).setUp()

        self.test_instance._is_open = True

    def test_returns_immediately_if_not_is_open(self):
        """ Should return immediately if is_open property is false. """
        instance = self.test_instance
        instance._is_open = False
        instance.pidfile = object()
        instance.close()
        self.assertFalse(self.mock_pidlockfile.__exit__.called)

    def test_exits_pidfile_context(self):
        """ Should exit the PID file context manager. """
        instance = self.test_instance
        instance.pidfile = self.mock_pidlockfile
        instance.close()
        self.mock_pidlockfile.__exit__.assert_called_with(None, None, None)

    def test_returns_none(self):
        """ Should return None. """
        instance = self.test_instance
        expected_result = None
        result = instance.close()
        self.assertIs(result, expected_result)

    def test_sets_is_open_false(self):
        """ Should set the `is_open` property to False. """
        instance = self.test_instance
        instance.close()
        self.assertEqual(False, instance.is_open)


@mock.patch.object(daemon.daemon.DaemonContext, "open")
class DaemonContext_context_manager_enter_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext.__enter__ method. """

    def test_opens_daemon_context(self, mock_func_daemoncontext_open):
        """ Should open the DaemonContext. """
        instance = self.test_instance
        instance.__enter__()
        mock_func_daemoncontext_open.assert_called_with()

    def test_returns_self_instance(self, mock_func_daemoncontext_open):
        """ Should return DaemonContext instance. """
        instance = self.test_instance
        expected_result = instance
        result = instance.__enter__()
        self.assertIs(result, expected_result)


@mock.patch.object(daemon.daemon.DaemonContext, "close")
class DaemonContext_context_manager_exit_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext.__exit__ method. """

    def setUp(self):
        """ Set up test fixtures. """
        super(DaemonContext_context_manager_exit_TestCase, self).setUp()

        self.test_args = dict(
                exc_type=object(),
                exc_value=object(),
                traceback=object(),
                )

    def test_closes_daemon_context(self, mock_func_daemoncontext_close):
        """ Should close the DaemonContext. """
        instance = self.test_instance
        args = self.test_args
        instance.__exit__(**args)
        mock_func_daemoncontext_close.assert_called_with()

    def test_returns_none(self, mock_func_daemoncontext_close):
        """ Should return None, indicating exception was not handled. """
        instance = self.test_instance
        args = self.test_args
        expected_result = None
        result = instance.__exit__(**args)
        self.assertIs(result, expected_result)


class DaemonContext_terminate_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext.terminate method. """

    def setUp(self):
        """ Set up test fixtures. """
        super(DaemonContext_terminate_TestCase, self).setUp()

        self.test_signal = signal.SIGTERM
        self.test_frame = None
        self.test_args = (self.test_signal, self.test_frame)

    def test_raises_system_exit(self):
        """ Should raise SystemExit. """
        instance = self.test_instance
        args = self.test_args
        expected_exception = SystemExit
        self.assertRaises(
                expected_exception,
                instance.terminate, *args)

    def test_exception_message_contains_signal_number(self):
        """ Should raise exception with a message containing signal number. """
        instance = self.test_instance
        args = self.test_args
        signal_number = self.test_signal
        expected_exception = SystemExit
        exc = self.assertRaises(
                expected_exception,
                instance.terminate, *args)
        self.assertIn(unicode(signal_number), unicode(exc))


class DaemonContext_get_exclude_file_descriptors_TestCase(
        DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext._get_exclude_file_descriptors function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(
                DaemonContext_get_exclude_file_descriptors_TestCase,
                self).setUp()

        self.test_files = {
                2: FakeFileDescriptorStringIO(),
                5: 5,
                11: FakeFileDescriptorStringIO(),
                17: None,
                23: FakeFileDescriptorStringIO(),
                37: 37,
                42: FakeFileDescriptorStringIO(),
                }
        for (fileno, item) in self.test_files.items():
            if hasattr(item, '_fileno'):
                item._fileno = fileno
        self.test_file_descriptors = set(
                fd for (fd, item) in self.test_files.items()
                if item is not None)
        self.test_file_descriptors.update(
                self.stream_files_by_name[name].fileno()
                for name in ['stdin', 'stdout', 'stderr']
                )

    def test_returns_expected_file_descriptors(self):
        """ Should return expected set of file descriptors. """
        instance = self.test_instance
        instance.files_preserve = list(self.test_files.values())
        expected_result = self.test_file_descriptors
        result = instance._get_exclude_file_descriptors()
        self.assertEqual(expected_result, result)

    def test_returns_stream_redirects_if_no_files_preserve(self):
        """ Should return only stream redirects if no files_preserve. """
        instance = self.test_instance
        instance.files_preserve = None
        expected_result = set(
                stream.fileno()
                for stream in self.stream_files_by_name.values())
        result = instance._get_exclude_file_descriptors()
        self.assertEqual(expected_result, result)

    def test_returns_empty_set_if_no_files(self):
        """ Should return empty set if no file options. """
        instance = self.test_instance
        for name in ['files_preserve', 'stdin', 'stdout', 'stderr']:
            setattr(instance, name, None)
        expected_result = set()
        result = instance._get_exclude_file_descriptors()
        self.assertEqual(expected_result, result)

    def test_omits_non_file_streams(self):
        """ Should omit non-file stream attributes. """
        instance = self.test_instance
        instance.files_preserve = list(self.test_files.values())
        stream_files = self.stream_files_by_name
        expected_result = self.test_file_descriptors.copy()
        for (pseudo_stream_name, pseudo_stream) in stream_files.items():
            test_non_file_object = object()
            setattr(instance, pseudo_stream_name, test_non_file_object)
            stream_fd = pseudo_stream.fileno()
            expected_result.discard(stream_fd)
        result = instance._get_exclude_file_descriptors()
        self.assertEqual(expected_result, result)

    def test_includes_verbatim_streams_without_file_descriptor(self):
        """ Should include verbatim any stream without a file descriptor. """
        instance = self.test_instance
        instance.files_preserve = list(self.test_files.values())
        stream_files = self.stream_files_by_name
        mock_fileno_method = mock.MagicMock(
                spec=sys.__stdin__.fileno,
                side_effect=ValueError)
        expected_result = self.test_file_descriptors.copy()
        for (pseudo_stream_name, pseudo_stream) in stream_files.items():
            test_non_fd_stream = StringIO()
            if not hasattr(test_non_fd_stream, 'fileno'):
                # Python < 3 StringIO doesn't have ‘fileno’ at all.
                # Add a method which raises an exception.
                test_non_fd_stream.fileno = mock_fileno_method
            setattr(instance, pseudo_stream_name, test_non_fd_stream)
            stream_fd = pseudo_stream.fileno()
            expected_result.discard(stream_fd)
            expected_result.add(test_non_fd_stream)
        result = instance._get_exclude_file_descriptors()
        self.assertEqual(expected_result, result)

    def test_omits_none_streams(self):
        """ Should omit any stream attribute which is None. """
        instance = self.test_instance
        instance.files_preserve = list(self.test_files.values())
        stream_files = self.stream_files_by_name
        expected_result = self.test_file_descriptors.copy()
        for (pseudo_stream_name, pseudo_stream) in stream_files.items():
            setattr(instance, pseudo_stream_name, None)
            stream_fd = pseudo_stream.fileno()
            expected_result.discard(stream_fd)
        result = instance._get_exclude_file_descriptors()
        self.assertEqual(expected_result, result)


class DaemonContext_make_signal_handler_TestCase(DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext._make_signal_handler function. """

    def test_returns_ignore_for_none(self):
        """ Should return SIG_IGN when None handler specified. """
        instance = self.test_instance
        target = None
        expected_result = signal.SIG_IGN
        result = instance._make_signal_handler(target)
        self.assertEqual(expected_result, result)

    def test_returns_method_for_name(self):
        """ Should return method of DaemonContext when name specified. """
        instance = self.test_instance
        target = 'terminate'
        expected_result = instance.terminate
        result = instance._make_signal_handler(target)
        self.assertEqual(expected_result, result)

    def test_raises_error_for_unknown_name(self):
        """ Should raise AttributeError for unknown method name. """
        instance = self.test_instance
        target = 'b0gUs'
        expected_error = AttributeError
        self.assertRaises(
                expected_error,
                instance._make_signal_handler, target)

    def test_returns_object_for_object(self):
        """ Should return same object for any other object. """
        instance = self.test_instance
        target = object()
        expected_result = target
        result = instance._make_signal_handler(target)
        self.assertEqual(expected_result, result)


class DaemonContext_make_signal_handler_map_TestCase(
        DaemonContext_BaseTestCase):
    """ Test cases for DaemonContext._make_signal_handler_map function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(DaemonContext_make_signal_handler_map_TestCase, self).setUp()

        self.test_instance.signal_map = {
                object(): object(),
                object(): object(),
                object(): object(),
                }

        self.test_signal_handlers = dict(
                (key, object())
                for key in self.test_instance.signal_map.values())
        self.test_signal_handler_map = dict(
                (key, self.test_signal_handlers[target])
                for (key, target) in self.test_instance.signal_map.items())

        def fake_make_signal_handler(target):
            return self.test_signal_handlers[target]

        func_patcher_make_signal_handler = mock.patch.object(
                daemon.daemon.DaemonContext, "_make_signal_handler",
                side_effect=fake_make_signal_handler)
        self.mock_func_make_signal_handler = (
                func_patcher_make_signal_handler.start())
        self.addCleanup(func_patcher_make_signal_handler.stop)

    def test_returns_constructed_signal_handler_items(self):
        """ Should return items as constructed via make_signal_handler. """
        instance = self.test_instance
        expected_result = self.test_signal_handler_map
        result = instance._make_signal_handler_map()
        self.assertEqual(expected_result, result)


try:
    FileNotFoundError
except NameError:
    # Python 2 uses IOError.
    FileNotFoundError = functools.partial(IOError, errno.ENOENT)


@mock.patch.object(os, "chdir")
class change_working_directory_TestCase(scaffold.TestCase):
    """ Test cases for change_working_directory function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(change_working_directory_TestCase, self).setUp()

        self.test_directory = object()
        self.test_args = dict(
                directory=self.test_directory,
                )

    def test_changes_working_directory_to_specified_directory(
            self,
            mock_func_os_chdir):
        """ Should change working directory to specified directory. """
        args = self.test_args
        directory = self.test_directory
        daemon.daemon.change_working_directory(**args)
        mock_func_os_chdir.assert_called_with(directory)

    def test_raises_daemon_error_on_os_error(
            self,
            mock_func_os_chdir):
        """ Should raise a DaemonError on receiving an IOError. """
        args = self.test_args
        test_error = FileNotFoundError("No such directory")
        mock_func_os_chdir.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_working_directory, **args)
        self.assertEqual(test_error, exc.__cause__)

    def test_error_message_contains_original_error_message(
            self,
            mock_func_os_chdir):
        """ Should raise a DaemonError with original message. """
        args = self.test_args
        test_error = FileNotFoundError("No such directory")
        mock_func_os_chdir.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_working_directory, **args)
        self.assertIn(unicode(test_error), unicode(exc))


@mock.patch.object(os, "chroot")
@mock.patch.object(os, "chdir")
class change_root_directory_TestCase(scaffold.TestCase):
    """ Test cases for change_root_directory function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(change_root_directory_TestCase, self).setUp()

        self.test_directory = object()
        self.test_args = dict(
                directory=self.test_directory,
                )

    def test_changes_working_directory_to_specified_directory(
            self,
            mock_func_os_chdir, mock_func_os_chroot):
        """ Should change working directory to specified directory. """
        args = self.test_args
        directory = self.test_directory
        daemon.daemon.change_root_directory(**args)
        mock_func_os_chdir.assert_called_with(directory)

    def test_changes_root_directory_to_specified_directory(
            self,
            mock_func_os_chdir, mock_func_os_chroot):
        """ Should change root directory to specified directory. """
        args = self.test_args
        directory = self.test_directory
        daemon.daemon.change_root_directory(**args)
        mock_func_os_chroot.assert_called_with(directory)

    def test_raises_daemon_error_on_os_error_from_chdir(
            self,
            mock_func_os_chdir, mock_func_os_chroot):
        """ Should raise a DaemonError on receiving an IOError from chdir. """
        args = self.test_args
        test_error = FileNotFoundError("No such directory")
        mock_func_os_chdir.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_root_directory, **args)
        self.assertEqual(test_error, exc.__cause__)

    def test_raises_daemon_error_on_os_error_from_chroot(
            self,
            mock_func_os_chdir, mock_func_os_chroot):
        """ Should raise a DaemonError on receiving an OSError from chroot. """
        args = self.test_args
        test_error = OSError(errno.EPERM, "No chroot for you!")
        mock_func_os_chroot.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_root_directory, **args)
        self.assertEqual(test_error, exc.__cause__)

    def test_error_message_contains_original_error_message(
            self,
            mock_func_os_chdir, mock_func_os_chroot):
        """ Should raise a DaemonError with original message. """
        args = self.test_args
        test_error = FileNotFoundError("No such directory")
        mock_func_os_chdir.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_root_directory, **args)
        self.assertIn(unicode(test_error), unicode(exc))


@mock.patch.object(os, "umask")
class change_file_creation_mask_TestCase(scaffold.TestCase):
    """ Test cases for change_file_creation_mask function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(change_file_creation_mask_TestCase, self).setUp()

        self.test_mask = object()
        self.test_args = dict(
                mask=self.test_mask,
                )

    def test_changes_umask_to_specified_mask(self, mock_func_os_umask):
        """ Should change working directory to specified directory. """
        args = self.test_args
        mask = self.test_mask
        daemon.daemon.change_file_creation_mask(**args)
        mock_func_os_umask.assert_called_with(mask)

    def test_raises_daemon_error_on_os_error_from_chdir(
            self,
            mock_func_os_umask):
        """ Should raise a DaemonError on receiving an OSError from umask. """
        args = self.test_args
        test_error = OSError(errno.EINVAL, "Whatchoo talkin' 'bout?")
        mock_func_os_umask.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_file_creation_mask, **args)
        self.assertEqual(test_error, exc.__cause__)

    def test_error_message_contains_original_error_message(
            self,
            mock_func_os_umask):
        """ Should raise a DaemonError with original message. """
        args = self.test_args
        test_error = FileNotFoundError("No such directory")
        mock_func_os_umask.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_file_creation_mask, **args)
        self.assertIn(unicode(test_error), unicode(exc))


@mock.patch.object(os, "setgid")
@mock.patch.object(os, "setuid")
class change_process_owner_TestCase(scaffold.TestCase):
    """ Test cases for change_process_owner function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(change_process_owner_TestCase, self).setUp()

        self.test_uid = object()
        self.test_gid = object()
        self.test_args = dict(
                uid=self.test_uid,
                gid=self.test_gid,
                )

    def test_changes_gid_and_uid_in_order(
            self,
            mock_func_os_setuid, mock_func_os_setgid):
        """ Should change process GID and UID in correct order.

            Since the process requires appropriate privilege to use
            either of `setuid` or `setgid`, changing the UID must be
            done last.

            """
        args = self.test_args
        daemon.daemon.change_process_owner(**args)
        mock_func_os_setuid.assert_called()
        mock_func_os_setgid.assert_called()

    def test_changes_group_id_to_gid(
            self,
            mock_func_os_setuid, mock_func_os_setgid):
        """ Should change process GID to specified value. """
        args = self.test_args
        gid = self.test_gid
        daemon.daemon.change_process_owner(**args)
        mock_func_os_setgid.assert_called(gid)

    def test_changes_user_id_to_uid(
            self,
            mock_func_os_setuid, mock_func_os_setgid):
        """ Should change process UID to specified value. """
        args = self.test_args
        uid = self.test_uid
        daemon.daemon.change_process_owner(**args)
        mock_func_os_setuid.assert_called(uid)

    def test_raises_daemon_error_on_os_error_from_setgid(
            self,
            mock_func_os_setuid, mock_func_os_setgid):
        """ Should raise a DaemonError on receiving an OSError from setgid. """
        args = self.test_args
        test_error = OSError(errno.EPERM, "No switching for you!")
        mock_func_os_setgid.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_process_owner, **args)
        self.assertEqual(test_error, exc.__cause__)

    def test_raises_daemon_error_on_os_error_from_setuid(
            self,
            mock_func_os_setuid, mock_func_os_setgid):
        """ Should raise a DaemonError on receiving an OSError from setuid. """
        args = self.test_args
        test_error = OSError(errno.EPERM, "No switching for you!")
        mock_func_os_setuid.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_process_owner, **args)
        self.assertEqual(test_error, exc.__cause__)

    def test_error_message_contains_original_error_message(
            self,
            mock_func_os_setuid, mock_func_os_setgid):
        """ Should raise a DaemonError with original message. """
        args = self.test_args
        test_error = OSError(errno.EINVAL, "Whatchoo talkin' 'bout?")
        mock_func_os_setuid.side_effect = test_error
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.change_process_owner, **args)
        self.assertIn(unicode(test_error), unicode(exc))


RLimitResult = collections.namedtuple('RLimitResult', ['soft', 'hard'])

fake_RLIMIT_CORE = object()

@mock.patch.object(resource, "RLIMIT_CORE", new=fake_RLIMIT_CORE)
@mock.patch.object(resource, "setrlimit", side_effect=(lambda x, y: None))
@mock.patch.object(resource, "getrlimit", side_effect=(lambda x: None))
class prevent_core_dump_TestCase(scaffold.TestCase):
    """ Test cases for prevent_core_dump function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(prevent_core_dump_TestCase, self).setUp()

    def test_sets_core_limit_to_zero(
            self,
            mock_func_resource_getrlimit, mock_func_resource_setrlimit):
        """ Should set the RLIMIT_CORE resource to zero. """
        expected_resource = fake_RLIMIT_CORE
        expected_limit = tuple(RLimitResult(soft=0, hard=0))
        daemon.daemon.prevent_core_dump()
        mock_func_resource_getrlimit.assert_called_with(expected_resource)
        mock_func_resource_setrlimit.assert_called_with(
                expected_resource, expected_limit)

    def test_raises_error_when_no_core_resource(
            self,
            mock_func_resource_getrlimit, mock_func_resource_setrlimit):
        """ Should raise DaemonError if no RLIMIT_CORE resource. """
        test_error = ValueError("Bogus platform doesn't have RLIMIT_CORE")
        def fake_getrlimit(res):
            if res == resource.RLIMIT_CORE:
                raise test_error
            else:
                return None
        mock_func_resource_getrlimit.side_effect = fake_getrlimit
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.prevent_core_dump)
        self.assertEqual(test_error, exc.__cause__)


@mock.patch.object(os, "close")
class close_file_descriptor_if_open_TestCase(scaffold.TestCase):
    """ Test cases for close_file_descriptor_if_open function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(close_file_descriptor_if_open_TestCase, self).setUp()

        self.fake_fd = 274

    def test_requests_file_descriptor_close(self, mock_func_os_close):
        """ Should request close of file descriptor. """
        fd = self.fake_fd
        daemon.daemon.close_file_descriptor_if_open(fd)
        mock_func_os_close.assert_called_with(fd)

    def test_ignores_badfd_error_on_close(self, mock_func_os_close):
        """ Should ignore OSError EBADF when closing. """
        fd = self.fake_fd
        test_error = OSError(errno.EBADF, "Bad file descriptor")
        def fake_os_close(fd):
            raise test_error
        mock_func_os_close.side_effect = fake_os_close
        daemon.daemon.close_file_descriptor_if_open(fd)
        mock_func_os_close.assert_called_with(fd)

    def test_raises_error_if_oserror_on_close(self, mock_func_os_close):
        """ Should raise DaemonError if an OSError occurs when closing. """
        fd = self.fake_fd
        test_error = OSError(object(), "Unexpected error")
        def fake_os_close(fd):
            raise test_error
        mock_func_os_close.side_effect = fake_os_close
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.close_file_descriptor_if_open, fd)
        self.assertEqual(test_error, exc.__cause__)

    def test_raises_error_if_ioerror_on_close(self, mock_func_os_close):
        """ Should raise DaemonError if an IOError occurs when closing. """
        fd = self.fake_fd
        test_error = IOError(object(), "Unexpected error")
        def fake_os_close(fd):
            raise test_error
        mock_func_os_close.side_effect = fake_os_close
        expected_error = daemon.daemon.DaemonOSEnvironmentError
        exc = self.assertRaises(
                expected_error,
                daemon.daemon.close_file_descriptor_if_open, fd)
        self.assertEqual(test_error, exc.__cause__)


class maxfd_TestCase(scaffold.TestCase):
    """ Test cases for module MAXFD constant. """

    def test_positive(self):
        """ Should be a positive number. """
        maxfd = daemon.daemon.MAXFD
        self.assertTrue(maxfd > 0)

    def test_integer(self):
        """ Should be an integer. """
        maxfd = daemon.daemon.MAXFD
        self.assertEqual(int(maxfd), maxfd)

    def test_reasonably_high(self):
        """ Should be reasonably high for default open files limit.

            If the system reports a limit of “infinity” on maximum
            file descriptors, we still need a finite number in order
            to close “all” of them. Ensure this is reasonably high
            to catch most use cases.

            """
        expected_minimum = 2048
        maxfd = daemon.daemon.MAXFD
        self.assertTrue(
                expected_minimum <= maxfd,
                msg=(
                    "MAXFD should be at least {minimum!r}"
                    " (got {maxfd!r})".format(
                        minimum=expected_minimum, maxfd=maxfd)))


fake_default_maxfd = 8
fake_RLIMIT_NOFILE = object()
fake_RLIM_INFINITY = object()
fake_rlimit_nofile_large = 2468

def fake_getrlimit_nofile_soft_infinity(resource):
    result = RLimitResult(soft=fake_RLIM_INFINITY, hard=object())
    if resource != fake_RLIMIT_NOFILE:
        result = NotImplemented
    return result

def fake_getrlimit_nofile_hard_infinity(resource):
    result = RLimitResult(soft=object(), hard=fake_RLIM_INFINITY)
    if resource != fake_RLIMIT_NOFILE:
        result = NotImplemented
    return result

def fake_getrlimit_nofile_hard_large(resource):
    result = RLimitResult(soft=object(), hard=fake_rlimit_nofile_large)
    if resource != fake_RLIMIT_NOFILE:
        result = NotImplemented
    return result

@mock.patch.object(daemon.daemon, "MAXFD", new=fake_default_maxfd)
@mock.patch.object(resource, "RLIMIT_NOFILE", new=fake_RLIMIT_NOFILE)
@mock.patch.object(resource, "RLIM_INFINITY", new=fake_RLIM_INFINITY)
@mock.patch.object(
        resource, "getrlimit",
        side_effect=fake_getrlimit_nofile_hard_large)
class get_maximum_file_descriptors_TestCase(scaffold.TestCase):
    """ Test cases for get_maximum_file_descriptors function. """

    def test_returns_system_hard_limit(self, mock_func_resource_getrlimit):
        """ Should return process hard limit on number of files. """
        expected_result = fake_rlimit_nofile_large
        result = daemon.daemon.get_maximum_file_descriptors()
        self.assertEqual(expected_result, result)

    def test_returns_module_default_if_hard_limit_infinity(
            self, mock_func_resource_getrlimit):
        """ Should return module MAXFD if hard limit is infinity. """
        mock_func_resource_getrlimit.side_effect = (
                fake_getrlimit_nofile_hard_infinity)
        expected_result = fake_default_maxfd
        result = daemon.daemon.get_maximum_file_descriptors()
        self.assertEqual(expected_result, result)


def fake_get_maximum_file_descriptors():
    return fake_default_maxfd

@mock.patch.object(resource, "RLIMIT_NOFILE", new=fake_RLIMIT_NOFILE)
@mock.patch.object(resource, "RLIM_INFINITY", new=fake_RLIM_INFINITY)
@mock.patch.object(
        resource, "getrlimit",
        new=fake_getrlimit_nofile_soft_infinity)
@mock.patch.object(
        daemon.daemon, "get_maximum_file_descriptors",
        new=fake_get_maximum_file_descriptors)
@mock.patch.object(daemon.daemon, "close_file_descriptor_if_open")
class close_all_open_files_TestCase(scaffold.TestCase):
    """ Test cases for close_all_open_files function. """

    def test_requests_all_open_files_to_close(
            self, mock_func_close_file_descriptor_if_open):
        """ Should request close of all open files. """
        expected_file_descriptors = range(fake_default_maxfd)
        expected_calls = [
                mock.call(fd) for fd in expected_file_descriptors]
        daemon.daemon.close_all_open_files()
        mock_func_close_file_descriptor_if_open.assert_has_calls(
                expected_calls, any_order=True)

    def test_requests_all_but_excluded_files_to_close(
            self, mock_func_close_file_descriptor_if_open):
        """ Should request close of all open files but those excluded. """
        test_exclude = set([3, 7])
        args = dict(
                exclude=test_exclude,
                )
        expected_file_descriptors = set(
                fd for fd in range(fake_default_maxfd)
                if fd not in test_exclude)
        expected_calls = [
                mock.call(fd) for fd in expected_file_descriptors]
        daemon.daemon.close_all_open_files(**args)
        mock_func_close_file_descriptor_if_open.assert_has_calls(
                expected_calls, any_order=True)


class detach_process_context_TestCase(scaffold.TestCase):
    """ Test cases for detach_process_context function. """

    class FakeOSExit(SystemExit):
        """ Fake exception raised for os._exit(). """

    def setUp(self):
        """ Set up test fixtures. """
        super(detach_process_context_TestCase, self).setUp()

        self.mock_module_os = mock.MagicMock(wraps=os)

        fake_pids = [0, 0]
        func_patcher_os_fork = mock.patch.object(
                os, "fork",
                side_effect=iter(fake_pids))
        self.mock_func_os_fork = func_patcher_os_fork.start()
        self.addCleanup(func_patcher_os_fork.stop)
        self.mock_module_os.attach_mock(self.mock_func_os_fork, "fork")

        func_patcher_os_setsid = mock.patch.object(os, "setsid")
        self.mock_func_os_setsid = func_patcher_os_setsid.start()
        self.addCleanup(func_patcher_os_setsid.stop)
        self.mock_module_os.attach_mock(self.mock_func_os_setsid, "setsid")

        def raise_os_exit(status=None):
            raise self.FakeOSExit(status)

        func_patcher_os_force_exit = mock.patch.object(
                os, "_exit",
                side_effect=raise_os_exit)
        self.mock_func_os_force_exit = func_patcher_os_force_exit.start()
        self.addCleanup(func_patcher_os_force_exit.stop)
        self.mock_module_os.attach_mock(self.mock_func_os_force_exit, "_exit")

    def test_parent_exits(self):
        """ Parent process should exit. """
        parent_pid = 23
        self.mock_func_os_fork.side_effect = iter([parent_pid])
        self.assertRaises(
                self.FakeOSExit,
                daemon.daemon.detach_process_context)
        self.mock_module_os.assert_has_calls([
                mock.call.fork(),
                mock.call._exit(0),
                ])

    def test_first_fork_error_raises_error(self):
        """ Error on first fork should raise DaemonProcessDetachError. """
        fork_errno = 13
        fork_strerror = "Bad stuff happened"
        test_error = OSError(fork_errno, fork_strerror)
        test_pids_iter = iter([test_error])

        def fake_fork():
            next_item = next(test_pids_iter)
            if isinstance(next_item, Exception):
                raise next_item
            else:
                return next_item

        self.mock_func_os_fork.side_effect = fake_fork
        exc = self.assertRaises(
                daemon.daemon.DaemonProcessDetachError,
                daemon.daemon.detach_process_context)
        self.assertEqual(test_error, exc.__cause__)
        self.mock_module_os.assert_has_calls([
                mock.call.fork(),
                ])

    def test_child_starts_new_process_group(self):
        """ Child should start new process group. """
        daemon.daemon.detach_process_context()
        self.mock_module_os.assert_has_calls([
                mock.call.fork(),
                mock.call.setsid(),
                ])

    def test_child_forks_next_parent_exits(self):
        """ Child should fork, then exit if parent. """
        fake_pids = [0, 42]
        self.mock_func_os_fork.side_effect = iter(fake_pids)
        self.assertRaises(
                self.FakeOSExit,
                daemon.daemon.detach_process_context)
        self.mock_module_os.assert_has_calls([
                mock.call.fork(),
                mock.call.setsid(),
                mock.call.fork(),
                mock.call._exit(0),
                ])

    def test_second_fork_error_reports_to_stderr(self):
        """ Error on second fork should cause report to stderr. """
        fork_errno = 17
        fork_strerror = "Nasty stuff happened"
        test_error = OSError(fork_errno, fork_strerror)
        test_pids_iter = iter([0, test_error])

        def fake_fork():
            next_item = next(test_pids_iter)
            if isinstance(next_item, Exception):
                raise next_item
            else:
                return next_item

        self.mock_func_os_fork.side_effect = fake_fork
        exc = self.assertRaises(
                daemon.daemon.DaemonProcessDetachError,
                daemon.daemon.detach_process_context)
        self.assertEqual(test_error, exc.__cause__)
        self.mock_module_os.assert_has_calls([
                mock.call.fork(),
                mock.call.setsid(),
                mock.call.fork(),
                ])

    def test_child_forks_next_child_continues(self):
        """ Child should fork, then continue if child. """
        daemon.daemon.detach_process_context()
        self.mock_module_os.assert_has_calls([
                mock.call.fork(),
                mock.call.setsid(),
                mock.call.fork(),
                ])


@mock.patch("os.getppid", return_value=765)
class is_process_started_by_init_TestCase(scaffold.TestCase):
    """ Test cases for is_process_started_by_init function. """

    def test_returns_false_by_default(self, mock_func_os_getppid):
        """ Should return False under normal circumstances. """
        expected_result = False
        result = daemon.daemon.is_process_started_by_init()
        self.assertIs(result, expected_result)

    def test_returns_true_if_parent_process_is_init(
            self, mock_func_os_getppid):
        """ Should return True if parent process is `init`. """
        init_pid = 1
        mock_func_os_getppid.return_value = init_pid
        expected_result = True
        result = daemon.daemon.is_process_started_by_init()
        self.assertIs(result, expected_result)


class is_socket_TestCase(scaffold.TestCase):
    """ Test cases for is_socket function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(is_socket_TestCase, self).setUp()

        def fake_getsockopt(level, optname, buflen=None):
            result = object()
            if optname is socket.SO_TYPE:
                result = socket.SOCK_RAW
            return result

        self.fake_socket_getsockopt_func = fake_getsockopt

        self.fake_socket_error = socket.error(
                errno.ENOTSOCK,
                "Socket operation on non-socket")

        self.mock_socket = mock.MagicMock(spec=socket.socket)
        self.mock_socket.getsockopt.side_effect = self.fake_socket_error

        def fake_socket_fromfd(fd, family, type, proto=None):
            return self.mock_socket

        func_patcher_socket_fromfd = mock.patch.object(
                socket, "fromfd",
                side_effect=fake_socket_fromfd)
        func_patcher_socket_fromfd.start()
        self.addCleanup(func_patcher_socket_fromfd.stop)

    def test_returns_false_by_default(self):
        """ Should return False under normal circumstances. """
        test_fd = 23
        expected_result = False
        result = daemon.daemon.is_socket(test_fd)
        self.assertIs(result, expected_result)

    def test_returns_true_if_stdin_is_socket(self):
        """ Should return True if `stdin` is a socket. """
        test_fd = 23
        getsockopt = self.mock_socket.getsockopt
        getsockopt.side_effect = self.fake_socket_getsockopt_func
        expected_result = True
        result = daemon.daemon.is_socket(test_fd)
        self.assertIs(result, expected_result)

    def test_returns_false_if_stdin_socket_raises_error(self):
        """ Should return True if `stdin` is a socket and raises error. """
        test_fd = 23
        getsockopt = self.mock_socket.getsockopt
        getsockopt.side_effect = socket.error(
                object(), "Weird socket stuff")
        expected_result = True
        result = daemon.daemon.is_socket(test_fd)
        self.assertIs(result, expected_result)


class is_process_started_by_superserver_TestCase(scaffold.TestCase):
    """ Test cases for is_process_started_by_superserver function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(is_process_started_by_superserver_TestCase, self).setUp()

        def fake_is_socket(fd):
            if sys.__stdin__.fileno() == fd:
                result = self.fake_stdin_is_socket_func()
            else:
                result = False
            return result

        self.fake_stdin_is_socket_func = (lambda: False)

        func_patcher_is_socket = mock.patch.object(
                daemon.daemon, "is_socket",
                side_effect=fake_is_socket)
        func_patcher_is_socket.start()
        self.addCleanup(func_patcher_is_socket.stop)

    def test_returns_false_by_default(self):
        """ Should return False under normal circumstances. """
        expected_result = False
        result = daemon.daemon.is_process_started_by_superserver()
        self.assertIs(result, expected_result)

    def test_returns_true_if_stdin_is_socket(self):
        """ Should return True if `stdin` is a socket. """
        self.fake_stdin_is_socket_func = (lambda: True)
        expected_result = True
        result = daemon.daemon.is_process_started_by_superserver()
        self.assertIs(result, expected_result)


@mock.patch.object(
        daemon.daemon, "is_process_started_by_superserver",
        return_value=False)
@mock.patch.object(
        daemon.daemon, "is_process_started_by_init",
        return_value=False)
class is_detach_process_context_required_TestCase(scaffold.TestCase):
    """ Test cases for is_detach_process_context_required function. """

    def test_returns_true_by_default(
            self,
            mock_func_is_process_started_by_init,
            mock_func_is_process_started_by_superserver):
        """ Should return True under normal circumstances. """
        expected_result = True
        result = daemon.daemon.is_detach_process_context_required()
        self.assertIs(result, expected_result)

    def test_returns_false_if_started_by_init(
            self,
            mock_func_is_process_started_by_init,
            mock_func_is_process_started_by_superserver):
        """ Should return False if current process started by init. """
        mock_func_is_process_started_by_init.return_value = True
        expected_result = False
        result = daemon.daemon.is_detach_process_context_required()
        self.assertIs(result, expected_result)

    def test_returns_true_if_started_by_superserver(
            self,
            mock_func_is_process_started_by_init,
            mock_func_is_process_started_by_superserver):
        """ Should return False if current process started by superserver. """
        mock_func_is_process_started_by_superserver.return_value = True
        expected_result = False
        result = daemon.daemon.is_detach_process_context_required()
        self.assertIs(result, expected_result)


def setup_streams_fixtures(testcase):
    """ Set up common test fixtures for standard streams. """
    testcase.stream_file_paths = dict(
            stdin=tempfile.mktemp(),
            stdout=tempfile.mktemp(),
            stderr=tempfile.mktemp(),
            )

    testcase.stream_files_by_name = dict(
            (name, FakeFileDescriptorStringIO())
            for name in ['stdin', 'stdout', 'stderr']
            )

    testcase.stream_files_by_path = dict(
            (testcase.stream_file_paths[name],
                testcase.stream_files_by_name[name])
            for name in ['stdin', 'stdout', 'stderr']
            )


@mock.patch.object(os, "dup2")
class redirect_stream_TestCase(scaffold.TestCase):
    """ Test cases for redirect_stream function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(redirect_stream_TestCase, self).setUp()

        self.test_system_stream = FakeFileDescriptorStringIO()
        self.test_target_stream = FakeFileDescriptorStringIO()
        self.test_null_file = FakeFileDescriptorStringIO()

        def fake_os_open(path, flag, mode=None):
            if path == os.devnull:
                result = self.test_null_file.fileno()
            else:
                raise FileNotFoundError("No such file", path)
            return result

        func_patcher_os_open = mock.patch.object(
                os, "open",
                side_effect=fake_os_open)
        self.mock_func_os_open = func_patcher_os_open.start()
        self.addCleanup(func_patcher_os_open.stop)

    def test_duplicates_target_file_descriptor(
            self, mock_func_os_dup2):
        """ Should duplicate file descriptor from target to system stream. """
        system_stream = self.test_system_stream
        system_fileno = system_stream.fileno()
        target_stream = self.test_target_stream
        target_fileno = target_stream.fileno()
        daemon.daemon.redirect_stream(system_stream, target_stream)
        mock_func_os_dup2.assert_called_with(target_fileno, system_fileno)

    def test_duplicates_null_file_descriptor_by_default(
            self, mock_func_os_dup2):
        """ Should by default duplicate the null file to the system stream. """
        system_stream = self.test_system_stream
        system_fileno = system_stream.fileno()
        target_stream = None
        null_path = os.devnull
        null_flag = os.O_RDWR
        null_file = self.test_null_file
        null_fileno = null_file.fileno()
        daemon.daemon.redirect_stream(system_stream, target_stream)
        self.mock_func_os_open.assert_called_with(null_path, null_flag)
        mock_func_os_dup2.assert_called_with(null_fileno, system_fileno)


class make_default_signal_map_TestCase(scaffold.TestCase):
    """ Test cases for make_default_signal_map function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(make_default_signal_map_TestCase, self).setUp()

        # Use whatever default string type this Python version needs.
        signal_module_name = str('signal')
        self.fake_signal_module = ModuleType(signal_module_name)

        fake_signal_names = [
                'SIGHUP',
                'SIGCLD',
                'SIGSEGV',
                'SIGTSTP',
                'SIGTTIN',
                'SIGTTOU',
                'SIGTERM',
                ]
        for name in fake_signal_names:
            setattr(self.fake_signal_module, name, object())

        module_patcher_signal = mock.patch.object(
                daemon.daemon, "signal", new=self.fake_signal_module)
        module_patcher_signal.start()
        self.addCleanup(module_patcher_signal.stop)

        default_signal_map_by_name = {
                'SIGTSTP': None,
                'SIGTTIN': None,
                'SIGTTOU': None,
                'SIGTERM': 'terminate',
                }
        self.default_signal_map = dict(
                (getattr(self.fake_signal_module, name), target)
                for (name, target) in default_signal_map_by_name.items())

    def test_returns_constructed_signal_map(self):
        """ Should return map per default. """
        expected_result = self.default_signal_map
        result = daemon.daemon.make_default_signal_map()
        self.assertEqual(expected_result, result)

    def test_returns_signal_map_with_only_ids_in_signal_module(self):
        """ Should return map with only signals in the `signal` module.

            The `signal` module is documented to only define those
            signals which exist on the running system. Therefore the
            default map should not contain any signals which are not
            defined in the `signal` module.

            """
        del(self.default_signal_map[self.fake_signal_module.SIGTTOU])
        del(self.fake_signal_module.SIGTTOU)
        expected_result = self.default_signal_map
        result = daemon.daemon.make_default_signal_map()
        self.assertEqual(expected_result, result)


@mock.patch.object(daemon.daemon.signal, "signal")
class set_signal_handlers_TestCase(scaffold.TestCase):
    """ Test cases for set_signal_handlers function. """

    def setUp(self):
        """ Set up test fixtures. """
        super(set_signal_handlers_TestCase, self).setUp()

        self.signal_handler_map = {
                signal.SIGQUIT: object(),
                signal.SIGSEGV: object(),
                signal.SIGINT: object(),
                }

    def test_sets_signal_handler_for_each_item(self, mock_func_signal_signal):
        """ Should set signal handler for each item in map. """
        signal_handler_map = self.signal_handler_map
        expected_calls = [
                mock.call(signal_number, handler)
                for (signal_number, handler) in signal_handler_map.items()]
        daemon.daemon.set_signal_handlers(signal_handler_map)
        self.assertEquals(expected_calls, mock_func_signal_signal.mock_calls)


@mock.patch.object(daemon.daemon.atexit, "register")
class register_atexit_function_TestCase(scaffold.TestCase):
    """ Test cases for register_atexit_function function. """

    def test_registers_function_for_atexit_processing(
            self, mock_func_atexit_register):
        """ Should register specified function for atexit processing. """
        func = object()
        daemon.daemon.register_atexit_function(func)
        mock_func_atexit_register.assert_called_with(func)


# Local variables:
# coding: utf-8
# mode: python
# End:
# vim: fileencoding=utf-8 filetype=python :