]>
Commit | Line | Data |
---|---|---|
fa39af54 | 1 | from __future__ import print_function |
e2dba705 CH |
2 | from datetime import datetime |
3 | import os | |
1a152698 | 4 | import requests |
1a152698 | 5 | import unittest |
1d6b70f9 | 6 | import sqlite3 |
7cbc5255 | 7 | import subprocess |
541bb91b CH |
8 | import sys |
9 | ||
10 | if sys.version_info[0] == 2: | |
11 | from urlparse import urljoin | |
12 | else: | |
13 | from urllib.parse import urljoin | |
14 | ||
7c876c30 | 15 | DAEMON = os.environ.get('DAEMON', 'authoritative') |
7cbc5255 | 16 | PDNSUTIL_CMD = os.environ.get('PDNSUTIL_CMD', 'NOT_SET BUT_THIS MIGHT_BE_A_LIST').split(' ') |
1d6b70f9 | 17 | SQLITE_DB = os.environ.get('SQLITE_DB', 'pdns.sqlite3') |
d19c22a1 CHB |
18 | SDIG = os.environ.get('SDIG', 'sdig') |
19 | DNSPORT = os.environ.get('DNSPORT', '53') | |
1a152698 CH |
20 | |
21 | class ApiTestCase(unittest.TestCase): | |
22 | ||
23 | def setUp(self): | |
24 | # TODO: config | |
825fa717 CH |
25 | self.server_address = '127.0.0.1' |
26 | self.server_port = int(os.environ.get('WEBPORT', '5580')) | |
27 | self.server_url = 'http://%s:%s/' % (self.server_address, self.server_port) | |
1a152698 | 28 | self.session = requests.Session() |
7f7481be | 29 | self.session.headers = {'X-API-Key': os.environ.get('APIKEY', 'changeme-key'), 'Origin': 'http://%s:%s' % (self.server_address, self.server_port)} |
1a152698 CH |
30 | |
31 | def url(self, relative_url): | |
541bb91b | 32 | return urljoin(self.server_url, relative_url) |
1a152698 | 33 | |
c1374bdb | 34 | def assert_success_json(self, result): |
fa75c441 CH |
35 | try: |
36 | result.raise_for_status() | |
37 | except: | |
541bb91b | 38 | print(result.content) |
fa75c441 | 39 | raise |
1a152698 | 40 | self.assertEquals(result.headers['Content-Type'], 'application/json') |
e2dba705 | 41 | |
ad918272 BZ |
42 | def assert_error_json(self, result): |
43 | self.assertTrue(400 <= result.status_code < 600, "Response has not an error code "+str(result.status_code)) | |
44 | self.assertEquals(result.headers['Content-Type'], 'application/json', "Response status code "+str(result.status_code)) | |
45 | ||
f0e76cee CH |
46 | def assert_success(self, result): |
47 | try: | |
48 | result.raise_for_status() | |
49 | except: | |
541bb91b | 50 | print(result.content) |
f0e76cee CH |
51 | raise |
52 | ||
e2dba705 CH |
53 | |
54 | def unique_zone_name(): | |
1d6b70f9 | 55 | return 'test-' + datetime.now().strftime('%d%H%S%M%f') + '.org.' |
7c876c30 | 56 | |
6eca6510 PL |
57 | def unique_tsigkey_name(): |
58 | return 'test-' + datetime.now().strftime('%d%H%S%M%f') + '-key' | |
7c876c30 | 59 | |
c1374bdb CH |
60 | def is_auth(): |
61 | return DAEMON == 'authoritative' | |
7c876c30 CH |
62 | |
63 | ||
c1374bdb CH |
64 | def is_recursor(): |
65 | return DAEMON == 'recursor' | |
1d6b70f9 CH |
66 | |
67 | ||
1d6b70f9 CH |
68 | def get_auth_db(): |
69 | """Return Connection to Authoritative backend DB.""" | |
70 | return sqlite3.Connection(SQLITE_DB) | |
71 | ||
72 | ||
73 | def get_db_records(zonename, qtype): | |
74 | with get_auth_db() as db: | |
75 | rows = db.execute(""" | |
b8cd24cc | 76 | SELECT name, type, content, ttl, ordername |
1d6b70f9 CH |
77 | FROM records |
78 | WHERE type = ? AND domain_id = ( | |
79 | SELECT id FROM domains WHERE name = ? | |
80 | )""", (qtype, zonename.rstrip('.'))).fetchall() | |
b8cd24cc | 81 | recs = [{'name': row[0], 'type': row[1], 'content': row[2], 'ttl': row[3], 'ordername': row[4]} for row in rows] |
541bb91b | 82 | print("DB Records:", recs) |
1d6b70f9 | 83 | return recs |
7cbc5255 CH |
84 | |
85 | ||
541bb91b CH |
86 | def pdnsutil(subcommand, *args): |
87 | try: | |
88 | return subprocess.check_output(PDNSUTIL_CMD + [subcommand] + list(args), close_fds=True).decode('ascii') | |
89 | except subprocess.CalledProcessError as except_inst: | |
7a0ea291 | 90 | raise RuntimeError("pdnsutil %s %s failed: %s" % (subcommand, args, except_inst.output.decode('ascii', errors='replace'))) |
541bb91b | 91 | |
7cbc5255 CH |
92 | def pdnsutil_rectify(zonename): |
93 | """Run pdnsutil rectify-zone on the given zone.""" | |
541bb91b | 94 | pdnsutil('rectify-zone', zonename) |
d19c22a1 CHB |
95 | |
96 | def sdig(*args): | |
97 | try: | |
98 | return subprocess.check_call([SDIG, '127.0.0.1', str(DNSPORT)] + list(args)) | |
99 | except subprocess.CalledProcessError as except_inst: | |
100 | raise RuntimeError("sdig %s %s failed: %s" % (command, args, except_inst.output.decode('ascii', errors='replace'))) | |
6eca6510 PL |
101 | |
102 | def get_db_tsigkeys(keyname): | |
103 | with get_auth_db() as db: | |
104 | rows = db.execute(""" | |
105 | SELECT name, algorithm, secret | |
106 | FROM tsigkeys | |
107 | WHERE name = ?""", (keyname, )).fetchall() | |
108 | keys = [{'name': row[0], 'algorithm': row[1], 'secret': row[2]} for row in rows] | |
fa39af54 | 109 | print("DB TSIG keys:", keys) |
6eca6510 PL |
110 | return keys |
111 |