python类patch()的实例源码

def test_build_error(experiment_class):
    with testing.postgresql.Postgresql() as postgresql:
        db_engine = create_engine(postgresql.url())
        ensure_db(db_engine)

        with TemporaryDirectory() as temp_dir:
            experiment = experiment_class(
                config=sample_config(),
                db_engine=db_engine,
                model_storage_class=FSModelStorageEngine,
                project_path=os.path.join(temp_dir, 'inspections'),
            )

            with mock.patch.object(experiment, 'build_matrices') as build_mock:
                build_mock.side_effect = RuntimeError('boom!')

                with pytest.raises(RuntimeError):
                    experiment()

def test_build_error_cleanup_timeout(_clean_up_mock, experiment_class):
    with testing.postgresql.Postgresql() as postgresql:
        db_engine = create_engine(postgresql.url())
        ensure_db(db_engine)

        with TemporaryDirectory() as temp_dir:
            experiment = experiment_class(
                config=sample_config(),
                db_engine=db_engine,
                model_storage_class=FSModelStorageEngine,
                project_path=os.path.join(temp_dir, 'inspections'),
                cleanup_timeout=0.02,  # Set short timeout
            )

            with mock.patch.object(experiment, 'build_matrices') as build_mock:
                build_mock.side_effect = RuntimeError('boom!')

                with pytest.raises(TimeoutError) as exc_info:
                    experiment()

    # Last exception is TimeoutError, but earlier error is preserved in
    # __context__, and will be noted as well in any standard traceback:
    assert exc_info.value.__context__ is build_mock.side_effect

def test_compare(self):
        q = quantizer.ImageQuantizer()

        qimages = q.quantize_multi([
            {'n_colors': 8, 'method': 'random'},
            {'n_colors': 16, 'method': 'random'},
            {'n_colors': 8, 'method': 'kmeans'},
            {'n_colors': 16, 'method': 'kmeans'},
            {'n_colors': 8, 'method': 'random+lab'},
            {'n_colors': 16, 'method': 'random+lab'},
            {'n_colors': 8, 'method': 'kmeans+lab'},
            {'n_colors': 16, 'method': 'kmeans+lab'},
        ], image_filename=self._get_image_path('Lenna.png'))

        with mock.patch('image_quantizer.quantizer.plt.show', lambda: None):
            quantizer.compare(*qimages)

def setUp(self):
        super().setUp()

        self.now = datetime(year=2017, month=5, day=12, hour=10, tzinfo=timezone.utc)

        self.timezone_patch = patch('valhalla.userrequests.contention.timezone')
        self.mock_timezone = self.timezone_patch.start()
        self.mock_timezone.now.return_value = self.now

        self.site_intervals_patch = patch('valhalla.userrequests.contention.get_site_rise_set_intervals')
        self.mock_site_intervals = self.site_intervals_patch.start()

        for i in range(24):
            request = mixer.blend(Request, state='PENDING')
            mixer.blend(
                Window, start=timezone.now(), end=timezone.now() + timedelta(hours=i), request=request
            )
            mixer.blend(
                Target, ra=random.randint(0, 360), dec=random.randint(-180, 180),
                proper_motion_ra=0.0, proper_motion_dec=0.0, type='SIDEREAL', request=request
            )
            mixer.blend(Molecule, instrument_name='1M0-SCICAM-SBIG', request=request)
            mixer.blend(Location, request=request)
            mixer.blend(Constraints, request=request)

def test_hcard_responds_on_404_on_unknown_user(self, client):
        response = client.get(reverse("federate:hcard", kwargs={"guid": "fehwuyfehiufhewiuhfiuhuiewfew"}))
        assert response.status_code == 404
        with patch("socialhome.federate.views.get_object_or_404") as mock_get:
            # Test also ValueError raising ending up as 404
            Profile.objects.filter(user__username="foobar").update(rsa_public_key="fooobar")
            mock_get.side_effect = ValueError()
            response = client.get(reverse("federate:hcard", kwargs={"guid": "foobar"}))
            assert response.status_code == 404

def test_pyembed_errors_swallowed(self):
        for error in [PyEmbedError, PyEmbedDiscoveryError, PyEmbedConsumerError, ValueError]:
            with patch("socialhome.content.previews.PyEmbed.embed", side_effect=error):
                result = fetch_oembed_preview(self.content, self.urls)
                self.assertFalse(result)

def test_define_services(uuid4):
    mock_service = mock.MagicMock()
    with mock.patch.dict(SERVICES, {'bar': mock_service}, clear=True):

        result = define_services([{'name': 'bar'}, {'name': 'baz'}])

    uuid4.assert_called_once_with()
    assert result == {uuid4().hex: mock_service.from_config.return_value}

def test_cli(argv):
    """test function."""
    with mock.patch('easysentiment.cli.scrape') as m_scrape, \
            mock.patch('easysentiment.cli.scrape_and_analyze') as m_saa, \
            mock.patch('easysentiment.cli.analyze_sentiment') as m_as:
        mock_dict = {
            'scrape': m_scrape,
            'scrape-and-analyze': m_saa,
            'analyze-sentiment': m_as,
        }
        from easysentiment.cli import cli
        cli([argv])
        mock_dict[argv].assert_called_once_with()

def test_build_version_with_valid_package_name(self):
        package = Mock(project_name='foo', version='1.0.0')
        setattr(settings, 'HEARTBEAT', {'package_name': 'foo'})
        with mock.patch.object(build.WorkingSet, 'find', return_value=package):
            distro = build.check(request=None)
            assert distro == {'name': 'foo', 'version': '1.0.0'}

def call_h1sync(*args, reports=None):
    if reports is None:
        reports = []
    with mock.patch('dashboard.h1.find_reports') as mock_find_reports:
        mock_find_reports.return_value = reports
        out = io.StringIO()
        call_command('h1sync', *args, stdout=out)
        return out.getvalue(), mock_find_reports

def test_default_security_groups(mock_properties, mock_details):
    """Make sure default Security Groups are added to the ingress rules."""
    ingress = {
        'test_app': [
            {
                'start_port': 30,
                'end_port': 30,
            },
        ],
    }

    mock_properties.return_value = {
        'security_group': {
            'ingress': ingress,
            'description': '',
        },
    }

    test_sg = {
        'myapp': [
            {
                'start_port': '22',
                'end_port': '22',
                'protocol': 'tcp'
            },
        ]
    }
    with mock.patch.dict('foremast.securitygroup.create_securitygroup.DEFAULT_SECURITYGROUP_RULES', test_sg):
        sg = SpinnakerSecurityGroup()
        ingress = sg.update_default_rules()
        assert 'myapp' in ingress

def test_write(self):
        """Test if ``WriteTensorBoard`` writes to its FileWriter."""
        hook = WriteTensorBoard(output_dir=self.tmpdir, model=self.get_model())
        with mock.patch.object(tf.summary.FileWriter, 'add_summary') as mocked_add_summary:
            hook.after_epoch(42, {})
            self.assertEqual(mocked_add_summary.call_count, 1)
            hook.after_epoch(43, {'valid': {'accuracy': 1.0}})
            self.assertEqual(mocked_add_summary.call_count, 2)
        hook.after_epoch(44, {'valid': {'accuracy': {'mean': np.float32(1.0)}}})
        hook.after_epoch(45, {'valid': {'accuracy': {'nanmean': 1.0}}})
        hook._summary_writer.close()

def test_image_variable(self):
        """Test if ``WriteTensorBoard`` checks the image variables properly."""
        hook = WriteTensorBoard(output_dir=self.tmpdir, model=self.get_model(), image_variables=['plot'])

        with mock.patch.dict('sys.modules', **{'cv2': cv2_mock}):
            with self.assertRaises(AssertionError):
                hook.after_epoch(0, {'train': {'plot': [None]}})

            with self.assertRaises(AssertionError):
                hook.after_epoch(1, {'train': {'plot': np.zeros((10,))}})

            hook.after_epoch(2, {'train': {'plot': np.zeros((10, 10, 3))}})
        hook._summary_writer.close()

def test_unknown_type(self):
        """Test if ``WriteTensorBoard`` handles unknown variable types as expected."""
        bad_epoch_data = {'valid': {'accuracy': 'bad_type'}}

        # test ignore
        hook = WriteTensorBoard(output_dir=self.tmpdir, model=self.get_model())
        with LogCapture(level=logging.INFO) as log_capture:
            hook.after_epoch(42, bad_epoch_data)
        log_capture.check()

        # test warn
        warn_hook = WriteTensorBoard(output_dir=self.tmpdir, model=self.get_model(), on_unknown_type='warn')
        with LogCapture(level=logging.INFO) as log_capture2:
            warn_hook.after_epoch(42, bad_epoch_data)
        log_capture2.check(('root', 'WARNING', 'Variable `accuracy` in stream `valid` has to be of type `int` '
                                               'or `float` (or a `dict` with a key named `mean` or `nanmean` '
                                               'whose corresponding value is of type `int` or `float`), '
                                               'found `<class \'str\'>` instead.'))

        # test error
        raise_hook = WriteTensorBoard(output_dir=self.tmpdir, model=self.get_model(), on_unknown_type='error')
        with self.assertRaises(ValueError):
            raise_hook.after_epoch(42, bad_epoch_data)

        with mock.patch.dict('sys.modules', **{'cv2': cv2_mock}):
            # test skip image variables
            skip_hook = WriteTensorBoard(output_dir=self.tmpdir, model=self.get_model(), on_unknown_type='error',
                                         image_variables=['accuracy'])
            skip_hook.after_epoch(42, {'valid': {'accuracy': np.zeros((10, 10, 3))}})
            skip_hook._summary_writer.close()

def testBigIno(self):
        """Test that index handles big inode numbers as found on Windows"""

        s = MagicMock()
        s.st_mode=33188
        s.st_ino=-5345198597064824875
        s.st_dev=65027
        s.st_nlink=1
        s.st_uid=1000
        s.st_gid=1000
        s.st_size=3
        s.st_atime=1452798827
        s.st_mtime=1452798827
        s.st_ctime=1452798827
        mock_lstat = MagicMock()
        mock_lstat.return_value = s

        with NamedTemporaryFile() as index:
            with TemporaryDirectory() as tmp:
                with open(os.path.join(tmp, "ghost"), 'wb') as f:
                    f.write(b'abc')

                with patch('os.lstat', mock_lstat):
                    hashDirectory(tmp, index.name)

                with open(index.name, "rb") as f:
                    assert f.read(4) == b'BOB1'