python类FakeLogger()的实例源码

def setUp(self):
        super(BaseTestCase, self).setUp()
        self.useFixture(fixtures.FakeLogger())

def setUp(self):
        super(TestCase, self).setUp()
        self.requests_mock = self.useFixture(fixture.Fixture())
        self.logger_mock = self.useFixture(fixtures.FakeLogger())

def setUp(self):
        super(BaseTestCase, self).setUp()
        self.useFixture(fixtures.FakeLogger())

        # If enabled, stdout and/or stderr is captured and will appear in
        # test results if that test fails.
        if strutils.bool_from_string(os.environ.get('OS_STDOUT_CAPTURE')):
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if strutils.bool_from_string(os.environ.get('OS_STDERR_CAPTURE')):
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))

def setUp(self):
        super(BaseTestCase, self).setUp()
        test_timeout = os.environ.get('OS_TEST_TIMEOUT', 30)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid, fail hard.
            print("OS_TEST_TIMEOUT set to invalid value"
                  " defaulting to no timeout")
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))

        if os.environ.get('OS_STDOUT_CAPTURE') in options.TRUE_VALUES:
            stdout = self.useFixture(fixtures.StringStream('stdout')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
        if os.environ.get('OS_STDERR_CAPTURE') in options.TRUE_VALUES:
            stderr = self.useFixture(fixtures.StringStream('stderr')).stream
            self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
        self.log_fixture = self.useFixture(
            fixtures.FakeLogger('pbr'))

        # Older git does not have config --local, so create a temporary home
        # directory to permit using git config --global without stepping on
        # developer configuration.
        self.useFixture(fixtures.TempHomeDir())
        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.FakeLogger())
        self.useFixture(fixtures.EnvironmentVariable('PBR_VERSION', '0.0'))

        self.temp_dir = self.useFixture(fixtures.TempDir()).path
        self.package_dir = os.path.join(self.temp_dir, 'testpackage')
        shutil.copytree(os.path.join(os.path.dirname(__file__), 'testpackage'),
                        self.package_dir)
        self.addCleanup(os.chdir, os.getcwd())
        os.chdir(self.package_dir)
        self.addCleanup(self._discard_testpackage)

def setUp(self):
        super(ConfigureLogging, self).setUp()
        if self.capture_logs:
            self.logger = self.useFixture(
                fixtures.FakeLogger(
                    format=self._format,
                    level=self.level,
                    nuke_handlers=True,
                )
            )
        else:
            logging.basicConfig(format=self._format, level=self.level)

def test_logs_on_missing_file(self):
        workdir = self.useFixture(fixtures.TempDir())
        fake_logger = self.useFixture(fixtures.FakeLogger())

        bad_path = os.path.join(workdir.path, 'path_does_not_exist')
        result = _build_doubles.extract_schemas_from_file(bad_path)

        self.assertIsNone(result)
        self.assertThat(
            fake_logger.output,
            Contains('Extracting schemas from %s' % bad_path))
        self.assertThat(
            fake_logger.output,
            Contains('Cannot extract schemas: No such file or directory'))

def test_logs_on_schema_extraction(self):
        workdir = self.useFixture(fixtures.TempDir())
        fake_logger = self.useFixture(fixtures.FakeLogger())

        good_path = os.path.join(workdir.path, 'my.py')
        with open(good_path, 'w') as f:
            f.write(dedent(
                """
                service = AcceptableService('vendor')

                root_api = service.api('/', 'root')

                @root_api.view(introduced_at='1.0')
                def my_view():
                    pass
                """))
        [schema] = _build_doubles.extract_schemas_from_file(good_path)

        self.assertEqual('root', schema.view_name)
        self.assertEqual('1.0', schema.version)
        self.assertEqual(['GET'], schema.methods)
        self.assertEqual(None, schema.input_schema)
        self.assertEqual(None, schema.output_schema)

        self.assertThat(
            fake_logger.output,
            Contains('Extracting schemas from %s' % good_path))
        self.assertThat(
            fake_logger.output,
            Contains('Extracted 1 schema'))


# To support testing, we need a version of ArgumentParser that doesn't call
# sys.exit on error, but rather throws an exception, so we can catch that in
# our tests:

def setUp(self):
        super(TestDrivers, self).setUp()
        self.logger = self.useFixture(
            fixtures.FakeLogger(
                format="%(levelname)8s [%(name)s] %(message)s",
                level=logging.DEBUG,
                nuke_handlers=True,
            )
        )

def test_output_not_containing_virtual_does_not_set_tag(self):
        logger = self.useFixture(FakeLogger())
        node = factory.make_Node()
        self.assertTagsEqual(node, [])
        set_virtual_tag(node, b"", 0)
        self.assertTagsEqual(node, [])
        self.assertIn(
            "No virtual type reported in VIRTUALITY_SCRIPT output for node "
            "%s" % node.system_id, logger.output)

def test_output_not_containing_virtual_does_not_remove_tag(self):
        logger = self.useFixture(FakeLogger())
        node = factory.make_Node()
        node.tags.add(self.getVirtualTag())
        self.assertTagsEqual(node, ["virtual"])
        set_virtual_tag(node, b"", 0)
        self.assertTagsEqual(node, ["virtual"])
        self.assertIn(
            "No virtual type reported in VIRTUALITY_SCRIPT output for node "
            "%s" % node.system_id, logger.output)

def test_hardware_updates_logs_invalid_xml(self):
        logger = self.useFixture(FakeLogger())
        update_hardware_details(factory.make_Node(), b"garbage", 0)
        expected_log = dedent("""\
        Invalid lshw data.
        Traceback (most recent call last):
        ...
        lxml.etree.XMLSyntaxError: Start tag expected, ...
        """)
        self.assertThat(
            logger.output, DocTestMatches(
                expected_log, self.doctest_flags))

def test_hardware_updates_does_nothing_when_exit_status_is_not_zero(self):
        logger = self.useFixture(FakeLogger(name='commissioningscript'))
        update_hardware_details(factory.make_Node(), b"garbage", exit_status=1)
        self.assertEqual("", logger.output)

def test_update_services_logs_when_service_not_recognised(self):
        service_name = factory.make_name("service")
        service = self.make_service(service_name)
        rack_controller = factory.make_RackController()
        with FakeLogger(services.__name__) as logger:
            update_services(rack_controller.system_id, [service])
        self.assertThat(logger.output, DocTestMatches(
            "Rack ... reported status for '...' but this is not a "
            "recognised service (status='...', info='...')."))

def test_logs_finding_existing_node_when_master(self):
        logger = self.useFixture(FakeLogger("maas"))
        node = factory.make_Node(node_type=NODE_TYPE.RACK_CONTROLLER)
        self.patch(rackcontrollers, "is_master_process").return_value = True
        register(system_id=node.system_id)
        self.assertEqual(
            "Existing rack controller '%s' running version 2.2 or below has "
            "connected to region '%s'." % (
                node.hostname, self.this_region.hostname),
            logger.output.strip())

def test_logs_finding_existing_node_when_master_with_version(self):
        logger = self.useFixture(FakeLogger("maas"))
        node = factory.make_Node(node_type=NODE_TYPE.RACK_CONTROLLER)
        self.patch(rackcontrollers, "is_master_process").return_value = True
        register(system_id=node.system_id, version="4.0")
        self.assertEqual(
            "Existing rack controller '%s' running version 4.0 has "
            "connected to region '%s'." % (
                node.hostname, self.this_region.hostname),
            logger.output.strip())

def test_logs_converting_region_controller(self):
        logger = self.useFixture(FakeLogger("maas"))
        node = factory.make_Node(node_type=NODE_TYPE.REGION_CONTROLLER)
        register(system_id=node.system_id)
        self.assertEqual(
            "Region controller '%s' running version 2.2 or below converted "
            "into a region and rack controller.\n" % node.hostname,
            logger.output)

def test_logs_converting_region_controller_with_version(self):
        logger = self.useFixture(FakeLogger("maas"))
        node = factory.make_Node(node_type=NODE_TYPE.REGION_CONTROLLER)
        register(system_id=node.system_id, version="7.8")
        self.assertEqual(
            "Region controller '%s' running version 7.8 converted "
            "into a region and rack controller.\n" % node.hostname,
            logger.output)

def test_logs_converting_existing_node(self):
        logger = self.useFixture(FakeLogger("maas"))
        node = factory.make_Node(node_type=NODE_TYPE.MACHINE)
        register(system_id=node.system_id)
        self.assertEqual(
            "Region controller '%s' converted '%s' running version 2.2 or "
            "below into a rack controller.\n" % (
                self.this_region.hostname, node.hostname),
            logger.output)

def test_logs_creating_new_rackcontroller(self):
        logger = self.useFixture(FakeLogger("maas"))
        hostname = factory.make_name("hostname")
        register(hostname=hostname)
        self.assertEqual(
            "New rack controller '%s' running version 2.2 or below was "
            "created by region '%s' upon first connection." % (
                hostname, self.this_region.hostname),
            logger.output.strip())

def test_logs_creating_new_rackcontroller_with_version(self):
        logger = self.useFixture(FakeLogger("maas"))
        hostname = factory.make_name("hostname")
        register(hostname=hostname, version="4.2")
        self.assertEqual(
            "New rack controller '%s' running version 4.2 was "
            "created by region '%s' upon first connection." % (
                hostname, self.this_region.hostname),
            logger.output.strip())