466 lines
23 KiB
Python
466 lines
23 KiB
Python
"""Offline safety and protocol regression tests; never contact a real device."""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from types import SimpleNamespace
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "scripts" / "device_adapter.py"
|
|
SN = "SYNTHETIC-TEST-SN-0001"
|
|
MODEL = "SYNTHETIC-MODEL"
|
|
TARGET = "9.8.7.6"
|
|
# RFC 5737 documentation address; every socket is replaced with an offline fixture.
|
|
TEST_IP = "192.0.2.10"
|
|
|
|
|
|
def response(sn=SN, model=MODEL, version=TARGET):
|
|
return {
|
|
"cmd": 30101, "session": 123, "code": 0,
|
|
"data": {
|
|
"sn": sn,
|
|
"inner": {"type": model, "version": version},
|
|
"outer": {"type": model, "version": "1.0.0.0"},
|
|
"hardware": {"type": model, "version": TARGET},
|
|
},
|
|
}
|
|
|
|
|
|
class FragmentedInfoSocket:
|
|
"""Transport-only double: creates a protocol response for the actual request."""
|
|
def __init__(self, reply):
|
|
self.reply = reply
|
|
self.pending = bytearray()
|
|
self.request = None
|
|
self.receives = 0
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def settimeout(self, _timeout):
|
|
pass
|
|
|
|
def sendall(self, packet):
|
|
assert packet[:4] == b"EPTP"
|
|
self.request = json.loads(packet[8:])
|
|
reply = dict(self.reply, session=self.request["session"])
|
|
payload = json.dumps(reply).encode("utf-8")
|
|
self.pending.extend(b"EPTP" + struct.pack("!I", len(payload)) + payload)
|
|
|
|
def recv(self, size):
|
|
self.receives += 1
|
|
count = min(size, 3, len(self.pending))
|
|
chunk = bytes(self.pending[:count])
|
|
del self.pending[:count]
|
|
return chunk
|
|
|
|
|
|
class UpgradeStatusSocket:
|
|
"""Run the real transfer state machine with controlled writes and status timing."""
|
|
def __init__(self, rom_size, success_after_wire_bytes):
|
|
self.success_after_wire_bytes = success_after_wire_bytes
|
|
payload = struct.pack("!BIIIII", 2, 123, rom_size, rom_size, rom_size, rom_size)
|
|
self.pending = bytearray(b"EPFP" + struct.pack("!II", 4001, len(payload)) + payload)
|
|
self.sent = bytearray()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def setblocking(self, _blocking):
|
|
pass
|
|
|
|
def send(self, packet):
|
|
count = min(len(packet), 7)
|
|
self.sent.extend(packet[:count])
|
|
return count
|
|
|
|
def recv(self, size):
|
|
count = min(size, 3, len(self.pending))
|
|
chunk = bytes(self.pending[:count])
|
|
del self.pending[:count]
|
|
return chunk
|
|
|
|
def ready(self, readable, writable, exceptional, timeout):
|
|
if len(self.sent) >= self.success_after_wire_bytes:
|
|
return ([self], [], [])
|
|
return ([], [self], [])
|
|
|
|
|
|
class DeviceAdapterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertTrue(SCRIPT.is_file(), "device_adapter.py has not been implemented")
|
|
spec = importlib.util.spec_from_file_location("device_adapter_under_test", SCRIPT)
|
|
self.adapter = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(self.adapter)
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.rom = Path(self.temp.name) / "synthetic.rom"
|
|
self.rom.write_bytes(b"synthetic firmware fixture\x00" * 30)
|
|
self.sha = hashlib.sha256(self.rom.read_bytes()).hexdigest()
|
|
|
|
def run_cli(self, argv, reply=None):
|
|
info_socket = FragmentedInfoSocket(reply or response())
|
|
connections = []
|
|
|
|
def connect(address, timeout):
|
|
connections.append(address)
|
|
if address[1] != 8942:
|
|
raise AssertionError("Unexpected write-port connection")
|
|
return info_socket
|
|
|
|
stdout = io.StringIO()
|
|
with patch.object(self.adapter.protocol.socket, "create_connection", side_effect=connect), contextlib.redirect_stdout(stdout):
|
|
code = self.adapter.main(argv)
|
|
return code, json.loads(stdout.getvalue()), connections, info_socket
|
|
|
|
def identity_args(self):
|
|
return ["--expected-sn", SN, "--expected-model", MODEL]
|
|
|
|
def upgrade_args(self):
|
|
return ["upgrade", "--host", TEST_IP, *self.identity_args(),
|
|
"--rom", str(self.rom), "--target-version", TARGET,
|
|
"--version-field", "inner.version"]
|
|
|
|
def run_transfer_cli(self, success_after_wire_bytes, host=TEST_IP):
|
|
info_socket = FragmentedInfoSocket(response())
|
|
transfer_socket = UpgradeStatusSocket(self.rom.stat().st_size, success_after_wire_bytes)
|
|
connections = []
|
|
|
|
def connect(address, timeout):
|
|
connections.append(address)
|
|
return info_socket if address[1] == 8942 else transfer_socket
|
|
|
|
stdout = io.StringIO()
|
|
args = self.upgrade_args() + ["--execute", "--rom-sha256", self.sha]
|
|
args[2] = host
|
|
with patch.object(self.adapter.protocol.socket, "create_connection", side_effect=connect), \
|
|
patch.object(self.adapter.protocol.select, "select", side_effect=transfer_socket.ready), \
|
|
contextlib.redirect_stdout(stdout):
|
|
code = self.adapter.main(args)
|
|
return code, json.loads(stdout.getvalue()), connections, transfer_socket
|
|
|
|
def test_v2_original_replaced_at_upgrade_connection_sends_approved_snapshot(self):
|
|
approved = self.rom.read_bytes()
|
|
for replacement in (b"B" * len(approved), b"C", b"D" * (len(approved) + 5)):
|
|
self.rom.write_bytes(approved)
|
|
info_socket = FragmentedInfoSocket(response())
|
|
transfer_socket = UpgradeStatusSocket(len(approved), 12 + len(approved))
|
|
def connect(address, timeout):
|
|
if address[1] == 8942:
|
|
return info_socket
|
|
self.rom.write_bytes(replacement)
|
|
return transfer_socket
|
|
out = io.StringIO()
|
|
with patch.object(self.adapter.protocol.socket, "create_connection", side_effect=connect), \
|
|
patch.object(self.adapter.protocol.select, "select", side_effect=transfer_socket.ready), \
|
|
contextlib.redirect_stdout(out):
|
|
code = self.adapter.main(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha])
|
|
self.assertEqual(code, 0, out.getvalue())
|
|
self.assertEqual(bytes(transfer_socket.sent)[12:], approved)
|
|
|
|
def verify_args(self, reply):
|
|
path = Path(self.temp.name) / "response.json"
|
|
path.write_text(json.dumps(reply), encoding="utf-8")
|
|
return ["verify-version", "--response", str(path), *self.identity_args(),
|
|
"--target-version", TARGET, "--version-field", "inner.version"]
|
|
|
|
def test_capabilities_is_offline_and_lists_explicit_v1_transfer(self):
|
|
stdout = io.StringIO()
|
|
with patch.object(self.adapter.protocol.socket, "socket", side_effect=AssertionError("Network forbidden")), \
|
|
patch.object(self.adapter.protocol.socket, "create_connection", side_effect=AssertionError("Network forbidden")), \
|
|
contextlib.redirect_stdout(stdout):
|
|
code = self.adapter.main(["capabilities"])
|
|
result = json.loads(stdout.getvalue())
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "CAPABILITIES")
|
|
self.assertFalse(result["network_used"])
|
|
self.assertFalse(result["firmware_verified"])
|
|
self.assertTrue(result["protocols"]["v1"]["implemented"])
|
|
self.assertEqual(result["protocols"]["v1"]["operations"], ["discover", "identify", "info", "upgrade", "verify-version"])
|
|
self.assertTrue(result["protocols"]["v1"]["upgrade_preflight"])
|
|
self.assertTrue(result["protocols"]["v1"]["upgrade_transfer"])
|
|
|
|
def test_v1_execute_requires_reviewed_digest_before_io_without_v2_fallback(self):
|
|
commands = [
|
|
self.upgrade_args() + ["--execute"],
|
|
]
|
|
for command in commands:
|
|
with self.subTest(command=command[0], execute="--execute" in command):
|
|
stdout = io.StringIO()
|
|
with patch.object(self.adapter, "load_vendor", side_effect=AssertionError("No fallback")), \
|
|
patch.object(self.adapter, "inspect_rom", side_effect=AssertionError("No ROM access")), \
|
|
patch.object(self.adapter, "read_response", side_effect=AssertionError("No response access")), \
|
|
patch.object(self.adapter.protocol, "query_device_info", side_effect=AssertionError("No V2 query")), \
|
|
patch.object(self.adapter.protocol, "transfer_rom", side_effect=AssertionError("No transfer")), \
|
|
patch.object(self.adapter.protocol.socket, "socket", side_effect=AssertionError("No UDP")), \
|
|
patch.object(self.adapter.protocol.socket, "create_connection", side_effect=AssertionError("No TCP")), \
|
|
contextlib.redirect_stdout(stdout):
|
|
code = self.adapter.main(command + ["--protocol", "v1"])
|
|
result = json.loads(stdout.getvalue())
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertIn("requires an explicitly reviewed", result["message"])
|
|
self.assertEqual(result["protocol"], "v1")
|
|
self.assertFalse(result["write_may_have_started"])
|
|
self.assertFalse(result["automatic_retry"])
|
|
self.assertEqual([e["event"] for e in result["events"]], ["upgrade_target_bound"])
|
|
|
|
def test_discover_preserves_native_arguments_with_default_or_explicit_v2(self):
|
|
native = ["--expected-sn", SN, "--expected-model", MODEL, "--timeout", "3"]
|
|
for selection in ([], ["--protocol", "v2"]):
|
|
with self.subTest(selection=selection):
|
|
discovery = SimpleNamespace(main=lambda arguments: (arguments, 17))
|
|
with patch.object(self.adapter, "load_vendor", return_value=discovery) as load:
|
|
result = self.adapter.main(["discover", *native, *selection])
|
|
self.assertEqual(result, (native, 17))
|
|
load.assert_called_once_with("v2_discover.py", "artemis_device_vendor_discover")
|
|
|
|
def test_explicit_v2_preflight_uses_only_identity_connection(self):
|
|
code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--protocol", "v2"])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["protocol"], "v2")
|
|
self.assertEqual(result["status"], "PREFLIGHT_ONLY")
|
|
self.assertFalse(result["write_executed"])
|
|
self.assertEqual(connections, [(TEST_IP, 8942)])
|
|
|
|
def test_info_needs_no_rom_and_reads_fragmented_response(self):
|
|
code, result, connections, sock = self.run_cli(["info", "--host", "synthetic.invalid", *self.identity_args()])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "IDENTITY_CONFIRMED")
|
|
self.assertEqual(result["response"]["data"]["sn"], SN)
|
|
self.assertEqual(connections, [("synthetic.invalid", 8942)])
|
|
self.assertGreater(sock.receives, 10)
|
|
self.assertEqual(sock.request["cmd"], 30101)
|
|
|
|
def test_v2_identify_observes_model_with_only_full_sn(self):
|
|
code, result, connections, _ = self.run_cli(["identify", "--host", TEST_IP, "--expected-sn", SN])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "IDENTITY_OBSERVED")
|
|
self.assertIn(MODEL, result["models"])
|
|
self.assertTrue(result["sn_verified"])
|
|
self.assertFalse(result["model_verified"])
|
|
self.assertFalse(result["write_executed"])
|
|
self.assertEqual(connections, [(TEST_IP, 8942)])
|
|
|
|
def test_v2_rejects_v1_native_version_fields_before_network(self):
|
|
for field in ("firmware", "dev_ver"):
|
|
with self.subTest(field=field):
|
|
args = self.upgrade_args()
|
|
args[-1] = field
|
|
code, result, connections, _ = self.run_cli(args)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_full_serial_mismatch_blocks_upgrade_before_write_port(self):
|
|
code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha], response(sn=SN + "-OTHER"))
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual([port for _, port in connections], [8942])
|
|
|
|
def test_model_mismatch_blocks_upgrade_before_write_port(self):
|
|
code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha], response(model="SYNTHETIC-OTHER-MODEL"))
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual([port for _, port in connections], [8942])
|
|
|
|
def test_upgrade_without_execute_never_connects_8943(self):
|
|
code, result, connections, _ = self.run_cli(self.upgrade_args())
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "PREFLIGHT_ONLY")
|
|
self.assertFalse(result["firmware_verified"])
|
|
self.assertEqual(result["rom"]["sha256"], self.sha)
|
|
self.assertEqual(result["target_version"], TARGET)
|
|
self.assertEqual(result["version_field"], "inner.version")
|
|
self.assertEqual([port for _, port in connections], [8942])
|
|
|
|
def test_upgrade_rejects_hostname_even_without_execute(self):
|
|
args = self.upgrade_args()
|
|
args[2] = "synthetic.invalid"
|
|
code, result, connections, _ = self.run_cli(args)
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_upgrade_rejects_determinable_non_unicast_targets(self):
|
|
for host in ("224.0.0.1", "0.0.0.0", "255.255.255.255", "::", "ff02::1",
|
|
"::ffff:0.0.0.0", "::ffff:224.0.0.1", "::ffff:255.255.255.255"):
|
|
with self.subTest(host=host):
|
|
args = self.upgrade_args()
|
|
args[2] = host
|
|
code, result, connections, _ = self.run_cli(args)
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_numeric_ip_is_bound_for_info_and_transfer_and_saved_as_evidence(self):
|
|
expected_wire_bytes = self.rom.stat().st_size + 12
|
|
for host, normalized in ((TEST_IP, TEST_IP), ("2001:0db8:0:0::10", "2001:db8::10")):
|
|
with self.subTest(host=host):
|
|
code, result, connections, sock = self.run_transfer_cli(expected_wire_bytes, host)
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result.get("bound_ip"), normalized)
|
|
self.assertEqual(connections, [(normalized, 8942), (normalized, 8943)])
|
|
self.assertEqual(bytes(sock.sent[:12]), b"EPFP" + struct.pack("!II", 4001, self.rom.stat().st_size))
|
|
self.assertEqual(bytes(sock.sent[12:]), self.rom.read_bytes())
|
|
self.assertEqual(result["transfer"]["bytes_sent"], self.rom.stat().st_size)
|
|
self.assertTrue(result["protocol_success"])
|
|
self.assertFalse(result["firmware_verified"])
|
|
|
|
def test_zero_or_partial_upload_status_2_cannot_claim_protocol_success(self):
|
|
for wire_bytes in (0, 12 + self.rom.stat().st_size // 2):
|
|
with self.subTest(wire_bytes=wire_bytes):
|
|
code, result, connections, sock = self.run_transfer_cli(wire_bytes)
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertFalse(result.get("protocol_success", False))
|
|
self.assertFalse(result["firmware_verified"])
|
|
self.assertTrue(result["write_may_have_started"])
|
|
self.assertTrue(result["postcheck_required"])
|
|
self.assertFalse(result["automatic_retry"])
|
|
self.assertEqual(connections, [(TEST_IP, 8942), (TEST_IP, 8943)])
|
|
self.assertLess(max(0, len(sock.sent) - 12), self.rom.stat().st_size)
|
|
self.assertEqual(result.get("bound_ip"), TEST_IP)
|
|
self.assertNotIn("protocol_success", [event["event"] for event in result["events"]])
|
|
|
|
def test_execute_requires_expected_rom_digest_before_network(self):
|
|
code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute"])
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_wrong_rom_digest_blocks_before_network(self):
|
|
code, _result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", "0" * 64])
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_offline_only_hashes_local_rom(self):
|
|
code, result, connections, _ = self.run_cli(["offline", "--rom", str(self.rom)])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "OFFLINE_READY")
|
|
self.assertEqual(result["rom"]["sha256"], self.sha)
|
|
self.assertFalse(result["firmware_verified"])
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_protocol_success_does_not_pass_firmware_acceptance(self):
|
|
with patch.object(self.adapter.protocol, "transfer_rom", return_value={"status": 2, "bytes_sent": self.rom.stat().st_size}) as transfer:
|
|
code, result, _connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(transfer.call_count, 1)
|
|
self.assertEqual(result["status"], "PROTOCOL_SUCCESS_POSTCHECK_REQUIRED")
|
|
self.assertFalse(result["firmware_verified"])
|
|
self.assertTrue(result["postcheck_required"])
|
|
|
|
def test_write_error_is_not_retried(self):
|
|
with patch.object(self.adapter.protocol, "transfer_rom", side_effect=OSError("synthetic disconnect")) as transfer:
|
|
code, result, _connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha])
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(transfer.call_count, 1)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertTrue(result.get("write_may_have_started"))
|
|
self.assertTrue(result.get("postcheck_required"))
|
|
|
|
def test_cli_configures_utf8_without_import_side_effects(self):
|
|
class ConfigStream(io.StringIO):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.configurations = []
|
|
|
|
def reconfigure(self, **kwargs):
|
|
self.configurations.append(kwargs)
|
|
|
|
stdout, stderr = ConfigStream(), ConfigStream()
|
|
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
|
code = self.adapter.main(["offline", "--rom", str(self.rom)])
|
|
self.assertEqual(code, 0)
|
|
self.assertIn({"encoding": "utf-8", "errors": "backslashreplace"}, stdout.configurations)
|
|
self.assertIn({"encoding": "utf-8", "errors": "backslashreplace"}, stderr.configurations)
|
|
|
|
def test_recorded_version_mismatch_cannot_pass(self):
|
|
code, result, connections, _ = self.run_cli(self.verify_args(response(version="0.0.0.0")))
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "FAIL")
|
|
self.assertFalse(result["firmware_verified"])
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_missing_firmware_version_cannot_pass_even_if_hardware_matches(self):
|
|
reply = response()
|
|
del reply["data"]["inner"]["version"]
|
|
code, result, _connections, _ = self.run_cli(self.verify_args(reply))
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "INCOMPLETE")
|
|
self.assertFalse(result["firmware_verified"])
|
|
|
|
def test_hardware_version_is_rejected_as_acceptance_field(self):
|
|
args = self.verify_args(response())
|
|
args[-1] = "hardware.version"
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as error:
|
|
self.adapter.main(args)
|
|
self.assertEqual(error.exception.code, 2)
|
|
|
|
def test_missing_data_cannot_pass(self):
|
|
code, result, _connections, _ = self.run_cli(self.verify_args({"cmd": 30101, "code": 0}))
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertFalse(result["firmware_verified"])
|
|
|
|
def test_wrong_identity_in_recorded_response_cannot_pass(self):
|
|
code, result, _connections, _ = self.run_cli(self.verify_args(response(sn="SYNTHETIC-WRONG-SN")))
|
|
self.assertNotEqual(code, 0)
|
|
self.assertFalse(result["firmware_verified"])
|
|
|
|
def test_matching_recorded_response_verifies_only_recorded_version(self):
|
|
code, result, connections, _ = self.run_cli(self.verify_args(response()))
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "PASS")
|
|
self.assertTrue(result["firmware_verified"])
|
|
self.assertEqual(result["evidence_source"], "recorded_response")
|
|
self.assertFalse(result["live_readback"])
|
|
self.assertEqual(connections, [])
|
|
|
|
def test_online_version_verification_reads_once(self):
|
|
args = ["verify-version", "--host", "synthetic.invalid", *self.identity_args(),
|
|
"--target-version", TARGET, "--version-field", "inner.version"]
|
|
code, result, connections, _ = self.run_cli(args)
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "PASS")
|
|
self.assertTrue(result["live_readback"])
|
|
self.assertEqual([port for _, port in connections], [8942])
|
|
|
|
def test_import_preserves_stdout_and_stderr_configuration(self):
|
|
class GuardStream(io.StringIO):
|
|
def reconfigure(self, **_kwargs):
|
|
raise AssertionError("Import must not reconfigure caller output")
|
|
|
|
# Reload the dependencies too: a cached import would hide the regression.
|
|
names = [name for name in sys.modules if name.startswith("artemis_device_vendor")]
|
|
removed = {name: sys.modules.pop(name) for name in names}
|
|
try:
|
|
with contextlib.redirect_stdout(GuardStream()), contextlib.redirect_stderr(GuardStream()):
|
|
spec = importlib.util.spec_from_file_location("safe_import_probe", SCRIPT)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
module.load_vendor("v2_discover.py", "artemis_device_vendor_discover")
|
|
finally:
|
|
sys.modules.update(removed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|