122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
"""Tests for the company/production-line license protocol."""
|
|
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
LICENSE_SPEC = importlib.util.spec_from_file_location(
|
|
"license_utils_under_test", ROOT / "license_utils.py"
|
|
)
|
|
LICENSE = importlib.util.module_from_spec(LICENSE_SPEC)
|
|
LICENSE_SPEC.loader.exec_module(LICENSE)
|
|
|
|
|
|
class FakePublicKey:
|
|
def verify(self, *args, **kwargs):
|
|
return None
|
|
|
|
|
|
def license_file(payload):
|
|
payload_b64 = base64.b64encode(json.dumps(payload).encode()).decode()
|
|
handle = tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False)
|
|
handle.write(f"{payload_b64}|{base64.b64encode(b'signature').decode()}")
|
|
handle.close()
|
|
return handle.name
|
|
|
|
|
|
NEW_LICENSE = {
|
|
"license_id": "license-123",
|
|
"customer": "Sample Co",
|
|
"company_id": "company-123",
|
|
"production_line_id": "line-123",
|
|
"device_id": "sample-co/line-1",
|
|
"issued": "2026-01-01 00:00",
|
|
"expiry": "2099-01-01 00:00",
|
|
"features": "*",
|
|
}
|
|
|
|
|
|
class LicenseProtocolTests(unittest.TestCase):
|
|
def verify_payload(self, payload):
|
|
path = license_file(payload)
|
|
self.addCleanup(os.unlink, path)
|
|
with patch.object(LICENSE, "_load_public_key", return_value=FakePublicKey()):
|
|
return LICENSE.verify_license(path)
|
|
|
|
def test_new_license_returns_company_and_line_identifiers(self):
|
|
self.assertEqual(self.verify_payload(NEW_LICENSE), NEW_LICENSE)
|
|
|
|
def test_old_license_remains_valid(self):
|
|
legacy = {
|
|
"customer": "Legacy Customer",
|
|
"issued": "2026-01-01",
|
|
"expiry": "2099-01-01",
|
|
"features": "*",
|
|
}
|
|
self.assertEqual(self.verify_payload(legacy), legacy)
|
|
|
|
def test_new_license_rejects_missing_organization_identifier(self):
|
|
invalid = dict(NEW_LICENSE)
|
|
invalid.pop("production_line_id")
|
|
with self.assertRaisesRegex(ValueError, "production_line_id"):
|
|
self.verify_payload(invalid)
|
|
|
|
def test_new_license_rejects_unsafe_device_id(self):
|
|
invalid = dict(NEW_LICENSE, device_id="sample-co/../line-1")
|
|
with self.assertRaisesRegex(ValueError, "device_id"):
|
|
self.verify_payload(invalid)
|
|
|
|
def test_online_active_status_is_accepted(self):
|
|
class Response:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"success": True, "valid": True, "status": "active",
|
|
"licenseId": NEW_LICENSE["license_id"]}
|
|
|
|
with patch.object(LICENSE.requests, "post", return_value=Response()):
|
|
LICENSE.validate_license_online(NEW_LICENSE)
|
|
|
|
def test_online_invalid_statuses_are_rejected(self):
|
|
for status in ("revoked", "expired", "device_mismatch"):
|
|
with self.subTest(status=status):
|
|
class Response:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"success": True, "valid": False, "status": status}
|
|
|
|
with patch.object(LICENSE.requests, "post", return_value=Response()):
|
|
with self.assertRaisesRegex(LICENSE.ExpiredError, status):
|
|
LICENSE.validate_license_online(NEW_LICENSE)
|
|
|
|
def test_online_network_failure_is_allowed_within_offline_grace(self):
|
|
LICENSE._last_online_success_monotonic = LICENSE._time_module.monotonic()
|
|
with patch.object(LICENSE.requests, "post",
|
|
side_effect=LICENSE.requests.ConnectionError("offline")):
|
|
LICENSE.validate_license_online(NEW_LICENSE)
|
|
|
|
def test_api_rejects_environment_device_id_mismatch(self):
|
|
fake_license_utils = types.ModuleType("license_utils")
|
|
fake_license_utils.get_verified_license = lambda: dict(NEW_LICENSE)
|
|
api_spec = importlib.util.spec_from_file_location("api_under_test", ROOT / "api.py")
|
|
api_module = importlib.util.module_from_spec(api_spec)
|
|
with patch.dict(os.environ, {"REINLOOP_DEVICE_ID": "other/line"}, clear=False), \
|
|
patch.dict(sys.modules, {"license_utils": fake_license_utils}):
|
|
with self.assertRaisesRegex(RuntimeError, "不一致"):
|
|
api_spec.loader.exec_module(api_module)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |