Last active
March 31, 2021 11:40
-
-
Save CodyKochmann/584e446dc94308add9dd32e75869fca0 to your computer and use it in GitHub Desktop.
This demonstrates how you can use custom functions in python to embed data validation for data types that are not currently supported by sqlite natively like ip addresses.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#/usr/bin/env python3 | |
# by: Cody Kochmann | |
# license: MIT | |
# last modified: 2021-03-31T5:39 | |
import ipaddress, sqlite3, sys, unittest | |
''' This demonstrates how you can use custom | |
functions in python to embed data validation | |
for data types that are not currently supported | |
by sqlite natively like ip addresses. | |
''' | |
def valid_ip(s: str) -> bool: | |
''' returns True if the input string is a valid ip address, otherwise False ''' | |
assert isinstance(s, str), s | |
print('validating:', s, file=sys.stderr) | |
try: | |
ipaddress.ip_address(s) | |
except: | |
print('invalid:', s, file=sys.stderr) | |
return False | |
else: | |
print('valid:', s, file=sys.stderr) | |
return True | |
class TestSQLiteIPValidation(unittest.TestCase): | |
def setUp(self): | |
# create an in memory db | |
self.db = sqlite3.connect(':memory:') | |
# add valid_ip as a function | |
self.db.create_function('valid_ip', 1, valid_ip) | |
# open a cursor for this test | |
self.cursor = self.db.cursor() | |
# create the schema utilizing the custom | |
# valid_ip function as a table assertion | |
self.cursor.execute(''' | |
CREATE TABLE hosts( | |
name TEXT NOT NULL, | |
ip TEXT NOT NULL, | |
CHECK (valid_ip(ip)) | |
); | |
''') | |
def tearDown(self): | |
# close this test's cursor | |
self.cursor.close() | |
# close this test's db | |
self.db.close() | |
def test_valid_ips(self): | |
# define test data | |
test_input = { | |
'gateway.home': '10.0.0.1', | |
'phone.home': '10.0.0.3' | |
} | |
# insert the test data | |
self.cursor.executemany( | |
'INSERT INTO hosts(name, ip) VALUES (?, ?)', | |
test_input.items() | |
) | |
# validate the test data is there | |
for host, ip in self.cursor.execute('SELECT * FROM hosts;'): | |
# test that the host was a test input | |
self.assertIn(host, test_input) | |
# test that the ip was the input value | |
self.assertEqual(ip, test_input[host]) | |
def test_invalid_ips(self): | |
# define test data | |
test_input = { | |
'gateway.home': '10.0.0.299', | |
'phone.home': '10.300.1.1' | |
} | |
# validate that sqlite raises an IntegrityError for the bad data | |
with self.assertRaises(sqlite3.IntegrityError): | |
# insert the test data | |
self.cursor.executemany( | |
'INSERT INTO hosts(name, ip) VALUES (?, ?)', | |
test_input.items() | |
) | |
# validate that no data was inserted | |
self.assertEqual( | |
self.cursor.execute('SELECT count(*) FROM hosts;').fetchone(), | |
(0,) | |
) | |
if __name__ == '__main__': | |
unittest.main(verbosity=2) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment