louis/ReinLoopTest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add auth check

Browse Source
This commit is contained in:
Epifnne
2026-08-03 11:16:49 +08:00
parent 6330b99860
commit 92780cedef
18 changed files with 381 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ReinLoop/tests/test_data_collector.py
+4
View File
@@ -15,6 +15,10 @@ def load_data_collector_module():
api = types.ModuleType("api")
api.base_url = "https://cloud.example"
api.data_record_url = "https://cloud.example"
api.device_post = lambda payload, timeout=10: {
"success": True,
"uploadMetadata": {"url": "https://upload.example"}
}
api.the_folder = "customer-a/line-1"
requests = types.ModuleType("requests")
ReinLoop/tests/test_device_heartbeat.py
+4 -18
View File
@@ -16,26 +16,12 @@ class DeviceHeartbeatTests(unittest.TestCase):
def test_sends_current_device_id_to_server(self):
calls = []
requests_module = types.ModuleType("requests")
class Response:
def raise_for_status(self):
return None
def json(self):
return {"success": True, "lastSeenAt": "2026-07-28T00:00:00.000Z"}
def post(url, json, timeout):
calls.append((url, json, timeout))
return Response()
requests_module.post = post
api_module = types.ModuleType("api")
api_module.data_record_url = "https://server.example"
api_module.the_folder = "company/line"
api_module.device_post = lambda payload, timeout=5: (
calls.append((payload, timeout)) or {"success": True, "lastSeenAt": "2026-07-28T00:00:00.000Z"}
)
with patch.dict(sys.modules, {"requests": requests_module, "api": api_module}):
timestamp = HEARTBEAT.heartbeat_device(timeout=7)
self.assertEqual(timestamp, "2026-07-28T00:00:00.000Z")
self.assertEqual(calls, [("https://server.example", {
"type": "deviceHeartbeat", "deviceId": "company/line"
}, 7)])
self.assertEqual(calls, [({"type": "deviceHeartbeat"}, 7)])
ReinLoop/tests/test_identification_config.py
+10 -34
View File
@@ -87,33 +87,23 @@ class IdentificationConfigTests(unittest.TestCase):
def test_download_requests_customer_config_and_validates_it(self):
calls = []
class FakeResponse:
def __init__(self, body=None, text=None):
self.body = body
self.text = text
def raise_for_status(self):
return None
def json(self):
return self.body
requests_module = types.ModuleType("requests")
requests_module.RequestException = Exception
def post(url, json, timeout):
calls.append(("post", url, json, timeout))
return FakeResponse({"success": True, "url": "https://temp/config"})
def device_post(payload, timeout):
calls.append(("device_post", payload, timeout))
return {"success": True, "url": "https://temp/config"}
def get(url, timeout):
calls.append(("get", url, timeout))
return FakeResponse(text=config_csv(VALID_CONFIG))
return types.SimpleNamespace(
text=config_csv(VALID_CONFIG),
raise_for_status=lambda: None
)
requests_module.post = post
requests_module.get = get
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
api_module.device_post = device_post
with patch.dict(sys.modules, {
"requests": requests_module,
@@ -122,28 +112,14 @@ class IdentificationConfigTests(unittest.TestCase):
result = download_identification_config(timeout=7)
self.assertEqual(result["repeat"], 2)
self.assertEqual(calls[0], (
"post",
"https://cloud/data_record",
{"type": "getIdentificationConfig", "deviceId": "客户A"},
7,
))
self.assertEqual(calls[0], ("device_post", {"type": "getIdentificationConfig"}, 7))
self.assertEqual(calls[1], ("get", "https://temp/config", 7))
def test_download_reports_cloud_rejection(self):
class FakeResponse:
def raise_for_status(self):
return None
def json(self):
return {"success": False, "errMsg": "配置不存在"}
requests_module = types.ModuleType("requests")
requests_module.RequestException = Exception
requests_module.post = lambda *args, **kwargs: FakeResponse()
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
api_module.device_post = lambda *args, **kwargs: {"success": False, "errMsg": "配置不存在"}
with patch.dict(sys.modules, {
"requests": requests_module,
ReinLoop/tests/test_volume_config.py
+19 -81
View File
@@ -57,28 +57,13 @@ class VolumeConfigTests(unittest.TestCase):
def test_customer_creates_exactly_one_request_instruction(self):
calls = []
class FakeResponse:
def raise_for_status(self):
return None
def json(self):
return {
"success": True,
"requestId": "request-1",
"expiresAtMs": 123456,
}
requests_module = types.ModuleType("requests")
def post(url, json, timeout):
calls.append((url, json, timeout))
return FakeResponse()
requests_module.post = post
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
api_module.device_post = lambda payload, timeout=10: (calls.append((payload, timeout)) or {
"success": True,
"requestId": "request-1",
"expiresAtMs": 123456,
})
with patch.dict(sys.modules, {
"requests": requests_module,
@@ -90,30 +75,16 @@ class VolumeConfigTests(unittest.TestCase):
"request_id": "request-1",
"expires_at_ms": 123456,
})
self.assertEqual(calls, [(
"https://cloud/data_record",
{"type": "createVolumeConfigRequest", "deviceId": "客户A"},
7,
)])
self.assertEqual(calls, [({"type": "createVolumeConfigRequest"}, 7)])
def test_pending_request_does_not_download_a_file(self):
calls = []
class FakeResponse:
def raise_for_status(self):
return None
def json(self):
return {"success": True, "ready": False, "expired": False}
requests_module = types.ModuleType("requests")
requests_module.post = lambda *args, **kwargs: (
calls.append(("post", kwargs["json"])) or FakeResponse()
)
requests_module.get = lambda *args, **kwargs: calls.append(("get", args[0]))
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
api_module.device_post = lambda payload, timeout=10: (
calls.append(("device_post", payload)) or {"success": True, "ready": False, "expired": False}
)
with patch.dict(sys.modules, {
"requests": requests_module,
@@ -122,36 +93,25 @@ class VolumeConfigTests(unittest.TestCase):
result = poll_volume_config_request("request-1")
self.assertEqual(result, {"ready": False, "expired": False})
self.assertEqual(calls, [("post", {
self.assertEqual(calls, [("device_post", {
"type": "getVolumeConfigRequest",
"deviceId": "客户A",
"requestId": "request-1",
})])
def test_ready_request_downloads_and_validates_json(self):
class FakeResponse:
def __init__(self, body):
self.body = body
def raise_for_status(self):
return None
def json(self):
return self.body
requests_module = types.ModuleType("requests")
requests_module.post = lambda *args, **kwargs: FakeResponse({
requests_module.post = lambda *args, **kwargs: types.SimpleNamespace()
api_module = types.ModuleType("api")
api_module.device_post = lambda payload, timeout=10: {
"success": True,
"ready": True,
"expired": False,
"url": "https://temp/volume.json",
})
requests_module.get = lambda *args, **kwargs: FakeResponse(
dict(VALID_CONFIG)
}
requests_module.get = lambda *args, **kwargs: types.SimpleNamespace(
raise_for_status=lambda: None,
json=lambda: dict(VALID_CONFIG)
)
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
with patch.dict(sys.modules, {
"requests": requests_module,
@@ -163,18 +123,9 @@ class VolumeConfigTests(unittest.TestCase):
self.assertEqual(result["config"], VALID_CONFIG)
def test_create_request_reports_server_rejection(self):
class FakeResponse:
def raise_for_status(self):
return None
def json(self):
return {"success": False, "errMsg": "尚未配置"}
requests_module = types.ModuleType("requests")
requests_module.post = lambda *args, **kwargs: FakeResponse()
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
api_module.device_post = lambda *args, **kwargs: {"success": False, "errMsg": "尚未配置"}
with patch.dict(sys.modules, {
"requests": requests_module,
@@ -185,21 +136,9 @@ class VolumeConfigTests(unittest.TestCase):
def test_acknowledges_the_same_request_for_cleanup(self):
calls = []
class FakeResponse:
def raise_for_status(self):
return None
def json(self):
return {"success": True, "deleted": 1}
requests_module = types.ModuleType("requests")
requests_module.post = lambda *args, **kwargs: (
calls.append(kwargs["json"]) or FakeResponse()
)
api_module = types.ModuleType("api")
api_module.data_record_url = "https://cloud/data_record"
api_module.the_folder = "客户A"
api_module.device_post = lambda payload, timeout=10: (calls.append(payload) or {"success": True, "deleted": 1})
with patch.dict(sys.modules, {
"requests": requests_module,
@@ -209,7 +148,6 @@ class VolumeConfigTests(unittest.TestCase):
self.assertEqual(calls, [{
"type": "ackVolumeConfigRequest",
"deviceId": "客户A",
"requestId": "request-1",
}])