From 5c23e6edb6a011efc55b797de67f89b0efc37fc4 Mon Sep 17 00:00:00 2001 From: Etienne Wodey Date: Wed, 16 Jun 2021 16:00:59 +0200 Subject: [PATCH] test: add regression tests for master.databases.DeviceDB Signed-off-by: Etienne Wodey --- artiq/test/test_device_db.py | 78 ++++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 artiq/test/test_device_db.py diff --git a/artiq/test/test_device_db.py b/artiq/test/test_device_db.py new file mode 100644 index 000000000..5cccaf379 --- /dev/null +++ b/artiq/test/test_device_db.py @@ -0,0 +1,78 @@ +"""Test device DB interface""" + +import unittest +import tempfile + +from artiq.master.databases import DeviceDB +from artiq.tools import file_import + + +DUMMY_DDB_FILE = """ +device_db = { + "core": { + "type": "local", + "module": "artiq.coredevice.core", + "class": "Core", + "arguments": {"host": "::1", "ref_period": 1e-09}, + }, + + "core-alias": "core", + "unresolved-alias": "dummy", +} +""" + + +class TestInvalidDeviceDB(unittest.TestCase): + def test_no_device_db_in_file(self): + with tempfile.NamedTemporaryFile(mode="w+", suffix=".py") as f: + print("", file=f, flush=True) + + with self.assertRaisesRegex(KeyError, "device_db"): + DeviceDB(f.name) + + +class TestDeviceDB(unittest.TestCase): + def setUp(self): + self.ddb_file = tempfile.NamedTemporaryFile(mode="w+", suffix=".py") + print(DUMMY_DDB_FILE, file=self.ddb_file, flush=True) + + self.ddb = DeviceDB(self.ddb_file.name) + + def test_get(self): + core = self.ddb.get("core") + self.assertEqual(core["class"], "Core") + + def test_get_alias(self): + with self.assertRaises(TypeError): # str indexing on str + self.ddb.get("core-alias")["class"] + + self.assertEqual( + self.ddb.get("core-alias", resolve_alias=True), self.ddb.get("core") + ) + + def test_get_unresolved_alias(self): + with self.assertRaisesRegex(KeyError, "dummy"): + self.ddb.get("unresolved-alias", resolve_alias=True) + + def test_update(self): + with self.assertRaises(KeyError): + self.ddb.get("core_log") + + update = """ +device_db["core_log"] = { + "type": "controller", + "host": "::1", + "port": 1068, + "command": "aqctl_corelog -p {port} --bind {bind} ::1", +}""" + + print(update, file=self.ddb_file, flush=True) + self.ddb.scan() + + self.assertEqual(self.ddb.get("core_log")["type"], "controller") + + def test_get_ddb(self): + ddb = self.ddb.get_device_db() + raw = file_import(self.ddb_file.name).device_db + + self.assertEqual(ddb, raw)