python类enable()的实例源码

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.GET, "http://test.gf/users",
            body=ujson.dumps([
                sam_profile,
                jack_profile,
            ]),
            content_type="application/json"
        )
        httpretty.register_uri(
            httpretty.GET, "http://test.gf/blog/sam",
            body=ujson.dumps(sams_articles),
            content_type="application/json"
        )
        httpretty.register_uri(
            httpretty.GET, "http://test.gf/blog/jack",
            body=ujson.dumps(jacks_articles),
            content_type="application/json"
        )

def testCanDeleteOwnedDomain(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.')

        url = reverse('domain-detail', args=(self.ownedDomains[1].pk,))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
        self.assertEqual(httpretty.last_request().method, 'DELETE')
        self.assertEqual(httpretty.last_request().headers['Host'], 'nsmaster:8081')

        httpretty.reset()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.')

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))

def testCanPostReverseDomains(self):
        name = '0.8.0.0.0.1.c.a.2.4.6.0.c.e.e.d.4.4.0.1.a.0.1.0.8.f.4.0.1.0.a.2.ip6.arpa'

        httpretty.enable()
        httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones', status=201)
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + name + '.',
                               body='{"rrsets": []}',
                               content_type="application/json")
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + name + './cryptokeys',
                               body='[]',
                               content_type="application/json")

        url = reverse('domain-list')
        data = {'name': name}
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(len(mail.outbox), 0)

def testCanDeleteOwnedDynDomain(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')

        url = reverse('domain-detail', args=(self.ownedDomains[1].pk,))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        # FIXME In this testing scenario, the parent domain dedyn.io does not
        # have the proper NS and DS records set up, so we cannot test their
        # deletion.

        httpretty.reset()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.')

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))

def setUp(self):
        if not hasattr(self, 'owner'):
            self.username = utils.generateRandomString(12)
            self.password = utils.generateRandomString(12)
            self.user = utils.createUser(self.username, self.password)
            self.token = utils.createToken(user=self.user)
            self.setCredentials(self.username, self.password)
            self.url = reverse('dyndns12update')

            self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
            self.domain = utils.generateDynDomainname()
            url = reverse('domain-list')
            data = {'name': self.domain}
            response = self.client.post(url, data)
            self.assertEqual(response.status_code, status.HTTP_201_CREATED)

            httpretty.enable()
            httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones')
            httpretty.register_uri(httpretty.GET,
                                   settings.NSLORD_PDNS_API + '/zones/' + self.domain + '.',
                                   body='{"rrsets": []}',
                                   content_type="application/json")
            httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.domain + '.')
            httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.domain + './notify')

def start_http_server(context):
    httpretty.disable()
    context.http_port = get_free_tcp_port()
    context.server = TornadoServer(context.http_port)
    context.server.start()
    ready = False
    timeout = 2
    started_at = time.time()
    while not ready:
        httpretty.disable()
        time.sleep(.1)
        try:
            requests.get('http://localhost:{0}/'.format(context.http_port))
            ready = True
        except:
            if time.time() - started_at >= timeout:
                break

    httpretty.enable()

def setup_httpretty():
    httpretty.enable()

def test_request_timeout():
    timeout = 0.1
    http_scheme = 'http'
    host = 'coordinator'
    port = 8080
    url = http_scheme + '://' + host + ':' + str(port) + constants.URL_STATEMENT_PATH

    def long_call(request, uri, headers):
        time.sleep(timeout * 2)
        return (200, headers, "delayed success")

    httpretty.enable()
    for method in [httpretty.POST, httpretty.GET]:
        httpretty.register_uri(method, url, body=long_call)

    # timeout without retry
    for request_timeout in [timeout, (timeout, timeout)]:
        req = PrestoRequest(
            host=host,
            port=port,
            user='test',
            http_scheme=http_scheme,
            max_attempts=1,
            request_timeout=request_timeout,
        )

        with pytest.raises(requests.exceptions.Timeout):
            req.get(url)

        with pytest.raises(requests.exceptions.Timeout):
            req.post('select 1')

    httpretty.disable()
    httpretty.reset()

def setUp(self):
        with open("line_file.json", "w") as f:
            f.write(JSON_LINE_CONTENT)
        with open("array_file.json", "w") as f:
            f.write(JSON_ARRAY_CONTENT)
        with open("object_file.json", "w") as f:
            f.write(JSON_OBJECT_CONTENT)
        with open("block_file.json", "w") as f:
            f.write(JSON_BLOCK_CONTENT)

        httpretty.enable()

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/line",
            body=JSON_LINE_CONTENT,
            content_type="text/plain"
        )

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/array",
            body=JSON_ARRAY_CONTENT,
            content_type="application/json"
        )

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/object",
            body=JSON_OBJECT_CONTENT,
            content_type="application/json"
        )

def setUp(self):
        self.records = [
            {"id": 1, "name": "Sam", "gender": 1},
            {"id": 2, "name": "Jack", "gender": 1},
            {"id": 3, "name": "Betty", "gender": 0},
        ]
        httpretty.enable()

def setUp(self):
        with open("test.csv", "w") as f:
            f.write(CSV_CONTENT)

        httpretty.enable()

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/csv",
            body=CSV_CONTENT,
            content_type="text/plain"
        )

def testCantDeleteOtherDomains(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.otherDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.otherDomains[1].name+ '.')

        url = reverse('domain-detail', args=(self.otherDomains[1].pk,))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
        self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))
        self.assertTrue(Domain.objects.filter(pk=self.otherDomains[1].pk).exists())

def testCanGetOwnedDomains(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + './cryptokeys',
                               body='[]',
                               content_type="application/json")

        url = reverse('domain-detail', args=(self.ownedDomains[1].pk,))
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['name'], self.ownedDomains[1].name)
        self.assertTrue(isinstance(response.data['keys'], list))

def testCantPostDomainAlreadyTakenInPdns(self):
        name = utils.generateDomainname()

        httpretty.enable()
        httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones',
                               body='{"error": "Domain \'' + name + '.\' already exists"}', status=422)

        url = reverse('domain-list')
        data = {'name': name}
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_409_CONFLICT)

def testRollback(self):
        name = utils.generateDomainname()

        httpretty.enable()
        httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones', body="some error", status=500)

        url = reverse('domain-list')
        data = {'name': name}
        self.client.post(url, data)

        self.assertFalse(Domain.objects.filter(name=name).exists())

def testCantDeleteOtherDynDomains(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.otherDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.otherDomains[1].name+ '.')

        url = reverse('domain-detail', args=(self.otherDomains[1].pk,))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
        self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))
        self.assertTrue(Domain.objects.filter(pk=self.otherDomains[1].pk).exists())

def testLimitDynDomains(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones')

        outboxlen = len(mail.outbox)

        url = reverse('domain-list')
        for i in range(settings.LIMIT_USER_DOMAIN_COUNT_DEFAULT-2):
            name = utils.generateDynDomainname()

            httpretty.register_uri(httpretty.GET,
                                   settings.NSLORD_PDNS_API + '/zones/' + name + '.',
                                   body='{"rrsets": []}',
                                   content_type="application/json")
            httpretty.register_uri(httpretty.GET,
                                   settings.NSLORD_PDNS_API + '/zones/' + name + './cryptokeys',
                                   body='[]',
                                   content_type="application/json")

            response = self.client.post(url, {'name': name})
            self.assertEqual(response.status_code, status.HTTP_201_CREATED)
            self.assertEqual(len(mail.outbox), outboxlen+i+1)

        data = {'name': utils.generateDynDomainname()}
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
        self.assertEqual(len(mail.outbox), outboxlen + settings.LIMIT_USER_DOMAIN_COUNT_DEFAULT-2)

def testDeleteCausesPdnsAPICall(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + './notify')

        # Create record, should cause a pdns PATCH request and a notify
        url = reverse('rrsets', args=(self.ownedDomains[1].name,))
        data = {'records': ['1.2.3.4'], 'ttl': 60, 'type': 'A'}
        self.client.post(url, json.dumps(data), content_type='application/json')

        # Delete record, should cause a pdns PATCH request and a notify
        url = reverse('rrset', args=(self.ownedDomains[1].name, '', 'A',))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        # Check pdns requests from creation
        result = json.loads(httpretty.httpretty.latest_requests[-4].parsed_body)
        self.assertEqual(result['rrsets'][0]['name'], self.ownedDomains[1].name + '.')
        self.assertEqual(result['rrsets'][0]['records'][0]['content'], '1.2.3.4')
        self.assertEqual(httpretty.httpretty.latest_requests[-3].method, 'PUT')

        # Check pdns requests from deletion
        result = json.loads(httpretty.httpretty.latest_requests[-2].parsed_body)
        self.assertEqual(result['rrsets'][0]['name'], self.ownedDomains[1].name + '.')
        self.assertEqual(result['rrsets'][0]['records'], [])
        self.assertEqual(httpretty.httpretty.latest_requests[-1].method, 'PUT')

def setUp(self):
        self.owner = utils.createUser()
        self.token = utils.createToken(user=self.owner)
        self.domain = utils.generateDynDomainname()
        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)

        url = reverse('domain-list')
        data = {'name': self.domain}
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.username = response.data['name']
        self.password = self.token
        self.client.credentials(HTTP_AUTHORIZATION='Basic ' + base64.b64encode((self.username + ':' + self.password).encode()).decode())

        httpretty.enable()
        httpretty.HTTPretty.allow_net_connect = False
        httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones')
        httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.domain + '.')
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + self.domain + '.',
                               body='{"rrsets": []}',
                               content_type="application/json")
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + self.domain + './cryptokeys',
                               body='[]',
                               content_type="application/json")
        httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.domain + './notify')

def _patched_get_content_and_type(self, url, harvest_job, page=1, content_type=None):

    httpretty.enable()

    value1, value2 = original_get_content_and_type(self, url, harvest_job, page, content_type)

    httpretty.disable()

    return value1, value2

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.POST, self.oauth2_client.ACCESS_TOKEN_URL,
            body=json.dumps(ACCESS_TOKEN),
            content_type='application/json')

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.POST, self.oauth2_client.ACCESS_TOKEN_URL,
            body=json.dumps(ACCESS_TOKEN),
            content_type='application/json')

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.POST, self.oauth2_client.ACCESS_TOKEN_URL,
            body=json.dumps(ACCESS_TOKEN),
            content_type='application/json')

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.GET, SERVICE_URL, body="LIVE")

def test_alive(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.GET, SERVICE_URL, body="LIVE")

        monitor = SocketPortMonitor(SERVICE_HOST, port=SERVICE_PORT);
        self.assertTrue(monitor.alive())

        httpretty.disable()
        httpretty.reset()

def test_check_alive(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.GET, SERVICE_URL, body="LIVE")

        self.assertTrue(check_alive('SocketPortMonitor', SERVICE_HOST, port='8888'))

        httpretty.disable()
        httpretty.reset()

def setUp(self):
        httpretty.enable()

def setup_function(function):
    httpretty.enable()
    register_collections_meta()
    register_collections()

def setup_function(function):
    httpretty.enable()
    register_collections_meta()
    register_collections()

def httpretty():
    httpretty_module.reset()
    httpretty_module.enable()
    yield httpretty_module
    httpretty_module.disable()