]>
git.ipfire.org Git - thirdparty/pdns.git/blob - regression-tests.api/test_helper.py
1 from datetime
import datetime
9 if sys
.version_info
[0] == 2:
10 from urlparse
import urljoin
12 from urllib
.parse
import urljoin
14 DAEMON
= os
.environ
.get('DAEMON', 'authoritative')
15 PDNSUTIL_CMD
= os
.environ
.get('PDNSUTIL_CMD', 'NOT_SET BUT_THIS MIGHT_BE_A_LIST').split(' ')
16 SQLITE_DB
= os
.environ
.get('SQLITE_DB', 'pdns.sqlite3')
17 SDIG
= os
.environ
.get('SDIG', 'sdig')
18 DNSPORT
= os
.environ
.get('DNSPORT', '53')
20 class ApiTestCase(unittest
.TestCase
):
24 self
.server_address
= '127.0.0.1'
25 self
.server_port
= int(os
.environ
.get('WEBPORT', '5580'))
26 self
.server_url
= 'http://%s:%s/' % (self
.server_address
, self
.server_port
)
27 self
.session
= requests
.Session()
28 self
.session
.headers
= {'X-API-Key': os
.environ
.get('APIKEY', 'changeme-key'), 'Origin': 'http://%s:%s' % (self
.server_address
, self
.server_port
)}
30 def url(self
, relative_url
):
31 return urljoin(self
.server_url
, relative_url
)
33 def assert_success_json(self
, result
):
35 result
.raise_for_status()
39 self
.assertEquals(result
.headers
['Content-Type'], 'application/json')
41 def assert_error_json(self
, result
):
42 self
.assertTrue(400 <= result
.status_code
< 600, "Response has not an error code "+str(result
.status_code
))
43 self
.assertEquals(result
.headers
['Content-Type'], 'application/json', "Response status code "+str(result
.status_code
))
45 def assert_success(self
, result
):
47 result
.raise_for_status()
53 def unique_zone_name():
54 return 'test-' + datetime
.now().strftime('%d%H%S%M%f') + '.org.'
58 return DAEMON
== 'authoritative'
62 return DAEMON
== 'recursor'
66 """Return Connection to Authoritative backend DB."""
67 return sqlite3
.Connection(SQLITE_DB
)
70 def get_db_records(zonename
, qtype
):
71 with
get_auth_db() as db
:
73 SELECT name, type, content, ttl
75 WHERE type = ? AND domain_id = (
76 SELECT id FROM domains WHERE name = ?
77 )""", (qtype
, zonename
.rstrip('.'))).fetchall()
78 recs
= [{'name': row
[0], 'type': row
[1], 'content': row
[2], 'ttl': row
[3]} for row
in rows
]
79 print("DB Records:", recs
)
83 def pdnsutil(subcommand
, *args
):
85 return subprocess
.check_output(PDNSUTIL_CMD
+ [subcommand
] + list(args
), close_fds
=True).decode('ascii')
86 except subprocess
.CalledProcessError
as except_inst
:
87 raise RuntimeError("pdnsutil %s %s failed: %s" % (command
, args
, except_inst
.output
.decode('ascii', errors
='replace')))
89 def pdnsutil_rectify(zonename
):
90 """Run pdnsutil rectify-zone on the given zone."""
91 pdnsutil('rectify-zone', zonename
)
95 return subprocess
.check_call([SDIG
, '127.0.0.1', str(DNSPORT
)] + list(args
))
96 except subprocess
.CalledProcessError
as except_inst
:
97 raise RuntimeError("sdig %s %s failed: %s" % (command
, args
, except_inst
.output
.decode('ascii', errors
='replace')))