|
|
@@ -0,0 +1,2513 @@
|
|
|
+"""Unit tests for OctoEverywhere's per-print AI detection integration."""
|
|
|
+
|
|
|
+import asyncio
|
|
|
+import logging
|
|
|
+from types import SimpleNamespace
|
|
|
+from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
+
|
|
|
+import httpx
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.services.octoeverywhere_detection import (
|
|
|
+ API_ERRORS,
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ ERROR_RETRY_INTERVAL,
|
|
|
+ IP_RESTRICTED_ERROR,
|
|
|
+ MAX_IMAGE_BYTES,
|
|
|
+ OctoEverywhereDetectionService,
|
|
|
+)
|
|
|
+
|
|
|
+MODULE = "backend.app.services.octoeverywhere_detection"
|
|
|
+FAKE_JPEG = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\xff\xd9"
|
|
|
+PRIMARY_URL = "https://gadget-region.octoeverywhere.com/api/gadget/v1/process/context-one"
|
|
|
+FALLBACK_URL = "https://gadget-pv1-oeapi.octoeverywhere.com/api/gadget/v1/process/context-one"
|
|
|
+USAGE_LIMIT_ERROR = "OE_FREE_USAGE_LIMIT_REACHED"
|
|
|
+
|
|
|
+
|
|
|
+def quota_response(retry_after=30):
|
|
|
+ return httpx.Response(
|
|
|
+ 429,
|
|
|
+ headers={"Retry-After": str(retry_after)},
|
|
|
+ json={"ErrorType": USAGE_LIMIT_ERROR, "ErrorDetails": "secret-api-key"},
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def ip_restricted_response():
|
|
|
+ return httpx.Response(403, json={"ErrorType": IP_RESTRICTED_ERROR, "ErrorDetails": "secret-api-key"})
|
|
|
+
|
|
|
+
|
|
|
+def context_payload(**overrides):
|
|
|
+ payload = {
|
|
|
+ "ContextId": "context-one",
|
|
|
+ "ProcessRequestUrl": PRIMARY_URL,
|
|
|
+ "FallbackProcessRequestUrl": FALLBACK_URL,
|
|
|
+ }
|
|
|
+ payload.update(overrides)
|
|
|
+ return payload
|
|
|
+
|
|
|
+
|
|
|
+def process_payload(**overrides):
|
|
|
+ payload = {
|
|
|
+ "NextProcessIntervalSec": {"Minimum": 40, "Recommended": 40},
|
|
|
+ "PrintQuality": 9,
|
|
|
+ "WarningSuggested": False,
|
|
|
+ "PauseSuggested": False,
|
|
|
+ "Score": 0,
|
|
|
+ }
|
|
|
+ payload.update(overrides)
|
|
|
+ return payload
|
|
|
+
|
|
|
+
|
|
|
+def settings(**overrides):
|
|
|
+ values = {
|
|
|
+ "enabled": True,
|
|
|
+ "api_key": "secret-api-key",
|
|
|
+ "confidence": "medium",
|
|
|
+ "action": "notify",
|
|
|
+ "enabled_printers": None,
|
|
|
+ "poll_interval": 20,
|
|
|
+ }
|
|
|
+ values.update(overrides)
|
|
|
+ return values
|
|
|
+
|
|
|
+
|
|
|
+def printer_status(**overrides):
|
|
|
+ values = {
|
|
|
+ "state": "RUNNING",
|
|
|
+ "subtask_name": "test-print",
|
|
|
+ "subtask_id": "print-1",
|
|
|
+ "gcode_file": "plate_1.gcode",
|
|
|
+ }
|
|
|
+ values.update(overrides)
|
|
|
+ return SimpleNamespace(**values)
|
|
|
+
|
|
|
+
|
|
|
+async def poll_once(detector, configuration):
|
|
|
+ """Wait for checks when a test needs the results of a complete polling cycle."""
|
|
|
+ await detector._poll_once(configuration)
|
|
|
+ await asyncio.gather(*detector._checks.values(), return_exceptions=True)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def detector():
|
|
|
+ service = OctoEverywhereDetectionService()
|
|
|
+ service._capture_frame = AsyncMock(return_value=FAKE_JPEG)
|
|
|
+ service._dispatch_action = AsyncMock()
|
|
|
+ return service
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def clock():
|
|
|
+ # Patch this module's time reference rather than the process-wide clock
|
|
|
+ # used by asyncio's own scheduler.
|
|
|
+ with patch(f"{MODULE}.time") as mocked:
|
|
|
+ mocked.monotonic.return_value = 100.0
|
|
|
+ yield mocked.monotonic
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def http_client():
|
|
|
+ with patch(f"{MODULE}.httpx.AsyncClient") as factory:
|
|
|
+ client = MagicMock()
|
|
|
+ client.post = AsyncMock()
|
|
|
+ client.__aenter__ = AsyncMock(return_value=client)
|
|
|
+ client.__aexit__ = AsyncMock(return_value=False)
|
|
|
+ factory.return_value = client
|
|
|
+ yield client
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def manager():
|
|
|
+ with patch("backend.app.services.printer_manager.printer_manager") as mocked:
|
|
|
+ mocked.is_connected.return_value = True
|
|
|
+ mocked.get_all_statuses.return_value = {1: printer_status()}
|
|
|
+ yield mocked
|
|
|
+
|
|
|
+
|
|
|
+class TestConnection:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "confidence, level", [("lowest", 1), ("low", 2), ("medium", 3), ("high", 4), ("highest", 5)]
|
|
|
+ )
|
|
|
+ async def test_creates_context_without_uploading_photo(self, detector, http_client, confidence, level):
|
|
|
+ http_client.post.return_value = httpx.Response(201, json=context_payload())
|
|
|
+
|
|
|
+ result = await detector.test_connection(" secret-api-key ", confidence)
|
|
|
+
|
|
|
+ assert result == {"ok": True, "status_code": 201, "error": None, "error_code": None}
|
|
|
+ http_client.post.assert_awaited_once_with(
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ headers={"X-API-Key": "secret-api-key"},
|
|
|
+ json={"WarningConfidenceLevel": level, "PauseConfidenceLevel": level},
|
|
|
+ )
|
|
|
+ detector._capture_frame.assert_not_awaited()
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+
|
|
|
+ async def test_empty_key_never_makes_a_request(self, detector, http_client):
|
|
|
+ result = await detector.test_connection(" ")
|
|
|
+ assert result["ok"] is False
|
|
|
+ assert "required" in result["error"]
|
|
|
+ assert result["error_code"] is None
|
|
|
+ http_client.post.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("field", ["ProcessRequestUrl", "FallbackProcessRequestUrl"])
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "url",
|
|
|
+ [
|
|
|
+ "http://gadget.octoeverywhere.com/process",
|
|
|
+ "https://octoeverywhere.com.attacker.example/process",
|
|
|
+ "https://attacker.example/process",
|
|
|
+ "https://gadget.octoeverywhere.com@attacker.example/process",
|
|
|
+ "https://user:password@gadget.octoeverywhere.com/process",
|
|
|
+ "https://gadget.octoeverywhere.com:444/process",
|
|
|
+ "https://127.0.0.1/process",
|
|
|
+ "https://gadget.octoeverywhere.com/process\n",
|
|
|
+ None,
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_rejects_untrusted_processing_urls(self, detector, http_client, field, url):
|
|
|
+ http_client.post.return_value = httpx.Response(200, json=context_payload(**{field: url}))
|
|
|
+ result = await detector.test_connection("secret-api-key")
|
|
|
+ assert result["ok"] is False
|
|
|
+ assert "invalid processing URL" in result["error"]
|
|
|
+ assert http_client.post.await_count == 1
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("payload", [None, [], {"ContextId": ""}, {"ErrorType": "OE_INVALID_API_KEY"}])
|
|
|
+ async def test_malformed_context_is_not_success(self, detector, http_client, payload):
|
|
|
+ http_client.post.return_value = httpx.Response(200, json=payload)
|
|
|
+ assert (await detector.test_connection("secret-api-key"))["ok"] is False
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "error_type, expected",
|
|
|
+ [
|
|
|
+ ("OE_INVALID_API_KEY", "rejected the Gadget API key"),
|
|
|
+ ("OE_API_KEY_DISABLED", "disabled"),
|
|
|
+ ("OE_API_KEY_BLOCKED_PAYMENT_FAILED", "Check the account status"),
|
|
|
+ (IP_RESTRICTED_ERROR, "already in use by another"),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_actionable_api_errors_do_not_echo_secrets(self, detector, http_client, error_type, expected):
|
|
|
+ http_client.post.return_value = httpx.Response(
|
|
|
+ 403, json={"ErrorType": error_type, "ErrorDetails": "secret-api-key"}
|
|
|
+ )
|
|
|
+ result = await detector.test_connection("secret-api-key")
|
|
|
+ assert result["ok"] is False
|
|
|
+ assert result["status_code"] == 403
|
|
|
+ assert expected in result["error"]
|
|
|
+ assert "secret-api-key" not in result["error"]
|
|
|
+ assert result["error_code"] == error_type
|
|
|
+
|
|
|
+ async def test_transport_errors_do_not_echo_secrets(self, detector, http_client):
|
|
|
+ http_client.post.side_effect = httpx.ConnectError("secret-api-key")
|
|
|
+ result = await detector.test_connection("secret-api-key")
|
|
|
+ assert result["ok"] is False
|
|
|
+ assert "secret-api-key" not in result["error"]
|
|
|
+ assert result["error_code"] is None
|
|
|
+
|
|
|
+ async def test_redirects_are_not_followed(self, detector, http_client):
|
|
|
+ http_client.post.return_value = httpx.Response(302, headers={"Location": "https://attacker.example"})
|
|
|
+ result = await detector.test_connection("secret-api-key")
|
|
|
+ assert result["ok"] is False
|
|
|
+ assert http_client.post.await_count == 1
|
|
|
+ from backend.app.services.octoeverywhere_detection import httpx as service_httpx
|
|
|
+
|
|
|
+ assert service_httpx.AsyncClient.call_args.kwargs["follow_redirects"] is False
|
|
|
+
|
|
|
+
|
|
|
+class TestProcessing:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "confidence, level", [("lowest", 1), ("low", 2), ("medium", 3), ("high", 4), ("highest", 5)]
|
|
|
+ )
|
|
|
+ async def test_inspection_context_uses_configured_confidence(self, detector, http_client, clock, confidence, level):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(confidence=confidence))
|
|
|
+ creation = http_client.post.await_args_list[0]
|
|
|
+ assert creation.args == (CREATE_CONTEXT_URL,)
|
|
|
+ assert creation.kwargs["json"] == {
|
|
|
+ "WarningConfidenceLevel": level,
|
|
|
+ "PauseConfidenceLevel": level,
|
|
|
+ }
|
|
|
+ assert detector.get_per_printer()[1]["frame_count"] == 1
|
|
|
+
|
|
|
+ async def test_wire_format_and_server_controlled_interval(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 75, "Recommended": 20})),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ status = printer_status()
|
|
|
+ await detector._check_printer(1, status, settings())
|
|
|
+
|
|
|
+ assert http_client.post.await_args.args == (PRIMARY_URL,)
|
|
|
+ assert http_client.post.await_args.kwargs == {
|
|
|
+ "headers": {"X-API-Key": "secret-api-key"},
|
|
|
+ "files": {"image": ("snapshot.jpg", FAKE_JPEG, "image/jpeg")},
|
|
|
+ }
|
|
|
+ assert detector.get_per_printer()[1] == {
|
|
|
+ "class": "safe",
|
|
|
+ "print_quality": 9,
|
|
|
+ "frame_count": 1,
|
|
|
+ "error": None,
|
|
|
+ "error_code": None,
|
|
|
+ }
|
|
|
+ clock.return_value = 174
|
|
|
+ await detector._check_printer(1, status, settings())
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ clock.return_value = 175
|
|
|
+ await detector._check_printer(1, status, settings())
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ assert detector._capture_frame.await_count == 2
|
|
|
+ assert detector.get_per_printer()[1]["frame_count"] == 2
|
|
|
+
|
|
|
+ async def test_interval_is_measured_after_response(self, detector, http_client, clock):
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ clock.return_value = 125
|
|
|
+ return httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 60, "Recommended": 20}))
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].next_check_at == 185
|
|
|
+
|
|
|
+ async def test_default_interval_is_twenty_seconds_when_server_allows_faster(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 20})),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].next_check_at == 120
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "fields",
|
|
|
+ [
|
|
|
+ {},
|
|
|
+ {"Recommended": 1},
|
|
|
+ {"Recommended": 5},
|
|
|
+ {"Recommended": 20},
|
|
|
+ {"Recommended": 300},
|
|
|
+ {"Recommended": None},
|
|
|
+ {"Recommended": True},
|
|
|
+ {"Recommended": "20"},
|
|
|
+ {"Recommended": {}},
|
|
|
+ {"Recommended": []},
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_recommended_interval_ignored_without_faster_inspection_hint(
|
|
|
+ self, detector, http_client, clock, fields
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, **fields})),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=12))
|
|
|
+ assert detector._states[1].interval == 5
|
|
|
+ assert detector._states[1].next_check_at == 112
|
|
|
+
|
|
|
+ async def test_latest_server_minimum_can_increase_and_decrease(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 20})),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 75, "Recommended": 90})),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 10, "Recommended": 20})),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 20})),
|
|
|
+ ]
|
|
|
+ configured = settings(poll_interval=20)
|
|
|
+ for now, minimum, next_check in [(100, 5, 120), (120, 75, 195), (195, 10, 215), (215, 5, 235)]:
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert detector._states[1].interval == minimum
|
|
|
+ assert detector._states[1].next_check_at == next_check
|
|
|
+ clock.return_value = next_check - 0.1
|
|
|
+ call_count = http_client.post.await_count
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert http_client.post.await_count == call_count
|
|
|
+ assert http_client.post.await_count == 5
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("poll_interval", [5, 20, 30])
|
|
|
+ async def test_first_inspection_is_immediate_with_configured_baseline(
|
|
|
+ self, detector, http_client, clock, poll_interval
|
|
|
+ ):
|
|
|
+ async def capture(printer_id):
|
|
|
+ assert detector._states[printer_id].interval is None
|
|
|
+ assert detector._states[printer_id].next_check_at == 100 + poll_interval
|
|
|
+ return FAKE_JPEG
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 20})),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=poll_interval))
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ assert detector._states[1].next_check_at == 100 + poll_interval
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "server_interval, poll_interval, expected",
|
|
|
+ [(5, 20, 20), (10, 5, 10), (40, 30, 40), (40, 20, 40), (5, 5, 5), (5, 21, 21), (180, 30, 180)],
|
|
|
+ )
|
|
|
+ async def test_inspection_interval_is_longer_of_setting_and_server_minimum(
|
|
|
+ self, detector, http_client, clock, server_interval, poll_interval, expected
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200, json=process_payload(NextProcessIntervalSec={"Minimum": server_interval, "Recommended": 20})
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 200, json=process_payload(NextProcessIntervalSec={"Minimum": server_interval, "Recommended": 20})
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ configured = settings(poll_interval=poll_interval)
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert detector._states[1].interval == server_interval
|
|
|
+ assert detector._states[1].next_check_at == 100 + expected
|
|
|
+ clock.return_value = 100 + expected - 0.1
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ clock.return_value = 100 + expected
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+
|
|
|
+ async def test_interval_edits_preserve_context_and_completed_actions(self, detector, http_client, manager, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(PauseSuggested=True, NextProcessIntervalSec={"Minimum": 10, "Recommended": 20}),
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(PauseSuggested=True, NextProcessIntervalSec={"Minimum": 10, "Recommended": 20}),
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(PauseSuggested=True, NextProcessIntervalSec={"Minimum": 10, "Recommended": 20}),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings(poll_interval=30))
|
|
|
+ context = detector._states[1].context
|
|
|
+ clock.return_value = 105
|
|
|
+ await poll_once(detector, settings(poll_interval=5))
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ assert detector._states[1].next_check_at == 110
|
|
|
+ clock.return_value = 110
|
|
|
+ await poll_once(detector, settings(poll_interval=5))
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ assert detector._states[1].next_check_at == 120
|
|
|
+ clock.return_value = 115
|
|
|
+ await poll_once(detector, settings(poll_interval=21))
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ assert detector._states[1].next_check_at == 131
|
|
|
+ clock.return_value = 131
|
|
|
+ await poll_once(detector, settings(poll_interval=21))
|
|
|
+ assert http_client.post.await_count == 4
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ detector._dispatch_action.assert_awaited_once()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("retry_after, expected", [(None, 460), ("300", 580)])
|
|
|
+ async def test_interval_edits_never_shorten_rate_limit_backoff(
|
|
|
+ self, detector, http_client, clock, retry_after, expected
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 180, "Recommended": 20})),
|
|
|
+ httpx.Response(
|
|
|
+ 429,
|
|
|
+ headers={"Retry-After": retry_after} if retry_after else {},
|
|
|
+ json={"ErrorType": "OE_CONTEXT_RATE_LIMITED"},
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=30))
|
|
|
+ clock.return_value = 280
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=30))
|
|
|
+ assert detector._states[1].next_check_at == expected
|
|
|
+ clock.return_value = expected - 1
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=5))
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+
|
|
|
+ async def test_raw_score_and_quality_never_trigger_actions(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(Score=100, PrintQuality=1)),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action="pause_and_off"))
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("action", ["notify", "pause", "pause_and_off"])
|
|
|
+ async def test_warning_then_pause_dispatches_each_action_once(self, detector, http_client, clock, action):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(WarningSuggested=True, PrintQuality=4)),
|
|
|
+ httpx.Response(200, json=process_payload(WarningSuggested=True, PrintQuality=3)),
|
|
|
+ httpx.Response(200, json=process_payload(WarningSuggested=True, PauseSuggested=True, PrintQuality=1)),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True, PrintQuality=1)),
|
|
|
+ ]
|
|
|
+ for now in (100, 140, 180, 220):
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action=action))
|
|
|
+
|
|
|
+ calls = detector._dispatch_action.await_args_list
|
|
|
+ assert calls[0].args == (1, "notify", "test-print", 4)
|
|
|
+ if action == "notify":
|
|
|
+ assert len(calls) == 1
|
|
|
+ else:
|
|
|
+ assert len(calls) == 2
|
|
|
+ assert calls[1].args == (1, action, "test-print", 1)
|
|
|
+ status = detector.get_status()
|
|
|
+ assert status["per_printer"][1]["class"] == "failure"
|
|
|
+ assert len(status["history"]) == 4
|
|
|
+ assert status["history"][0]["print_quality"] == 1
|
|
|
+ assert status["history"][0]["timestamp"].endswith("+00:00")
|
|
|
+
|
|
|
+ async def test_pause_without_warning_notifies_once(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True, PrintQuality=2)),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action="pause"))
|
|
|
+ detector._dispatch_action.assert_awaited_once_with(1, "pause", "test-print", 2)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("action", ["pause", "pause_and_off"])
|
|
|
+ async def test_pause_after_warning_sends_its_own_notification(self, detector, http_client, clock, action):
|
|
|
+ # Run the real dispatcher: the user must hear that the printer stopped,
|
|
|
+ # not only about the warning that came before it.
|
|
|
+ detector._dispatch_action = OctoEverywhereDetectionService._dispatch_action.__get__(detector)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(WarningSuggested=True, PrintQuality=4)),
|
|
|
+ httpx.Response(200, json=process_payload(WarningSuggested=True, PauseSuggested=True, PrintQuality=1)),
|
|
|
+ ]
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.obico_actions._get_printer_name", return_value="Printer"),
|
|
|
+ patch("backend.app.services.obico_actions._pause_print") as pause,
|
|
|
+ patch("backend.app.services.obico_actions._turn_off_linked_plugs", new_callable=AsyncMock),
|
|
|
+ patch("backend.app.services.obico_actions._notify", new_callable=AsyncMock) as notify,
|
|
|
+ ):
|
|
|
+ for now in (100, 140):
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action=action))
|
|
|
+ pause.assert_called_once_with(1)
|
|
|
+ assert [call.args[4] for call in notify.await_args_list] == ["notify", action]
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "fields",
|
|
|
+ [
|
|
|
+ {"PrintQuality": True},
|
|
|
+ {"PrintQuality": 0},
|
|
|
+ {"PrintQuality": 11},
|
|
|
+ {"PrintQuality": None},
|
|
|
+ {"PauseSuggested": "false"},
|
|
|
+ {"WarningSuggested": 1},
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_invalid_assessment_is_an_error_and_never_acts(self, detector, http_client, clock, fields):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(**fields)),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ result = detector.get_per_printer()[1]
|
|
|
+ assert result["class"] == "error"
|
|
|
+ assert result["frame_count"] == 0
|
|
|
+ assert result["print_quality"] is None
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "fields",
|
|
|
+ [
|
|
|
+ {},
|
|
|
+ {"NextProcessIntervalSec": None},
|
|
|
+ {"NextProcessIntervalSec": True},
|
|
|
+ {"NextProcessIntervalSec": 20},
|
|
|
+ {"NextProcessIntervalSec": "20"},
|
|
|
+ {"NextProcessIntervalSec": []},
|
|
|
+ {"NextProcessIntervalSec": {}},
|
|
|
+ {"NextProcessIntervalSec": {"Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": None, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": True, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": False, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": 0, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": -1, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": 5.5, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": "5", "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": {}, "Recommended": 20}},
|
|
|
+ {"NextProcessIntervalSec": {"Minimum": [], "Recommended": 20}},
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_invalid_interval_retries_without_actions(self, detector, http_client, clock, fields):
|
|
|
+ payload = process_payload(PauseSuggested=True)
|
|
|
+ payload.pop("NextProcessIntervalSec")
|
|
|
+ payload.update(fields)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=payload),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 20})),
|
|
|
+ ]
|
|
|
+ configured = settings(poll_interval=5, action="pause_and_off")
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert "invalid processing interval" in state.error
|
|
|
+ assert state.frame_count == 0
|
|
|
+ assert state.interval is None
|
|
|
+ assert state.next_check_at == 100 + ERROR_RETRY_INTERVAL
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ clock.return_value = 100 + ERROR_RETRY_INTERVAL - 0.1
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ clock.return_value = 100 + ERROR_RETRY_INTERVAL
|
|
|
+ await detector._check_printer(1, printer_status(), configured)
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ assert state.context is context
|
|
|
+ assert state.error is None
|
|
|
+ assert state.frame_count == 1
|
|
|
+ assert state.next_check_at == 100 + ERROR_RETRY_INTERVAL + 5
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_failure_uses_fallback_after_backoff_and_keeps_it(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.ConnectError("secret-api-key"),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert "secret-api-key" not in detector.get_status()["last_error"]
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ clock.return_value = 100 + ERROR_RETRY_INTERVAL - 1
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ for now in (160, 200):
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ FALLBACK_URL,
|
|
|
+ FALLBACK_URL,
|
|
|
+ ]
|
|
|
+ assert detector.get_status()["last_error"] is None
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+
|
|
|
+ async def test_retry_after_and_previous_server_interval_are_honored(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 180, "Recommended": 20})),
|
|
|
+ httpx.Response(429, headers={"Retry-After": "300"}, json={"ErrorType": "OE_CONTEXT_RATE_LIMITED"}),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ clock.return_value = 280
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].next_check_at == 580
|
|
|
+ assert detector._states[1].frame_count == 1
|
|
|
+
|
|
|
+ async def test_invalid_assessment_still_honors_valid_long_interval(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200, json=process_payload(NextProcessIntervalSec={"Minimum": 600, "Recommended": 20}, PrintQuality=None)
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].next_check_at == 700
|
|
|
+
|
|
|
+ async def test_failed_capture_does_not_upload_and_backs_off(self, detector, http_client, clock):
|
|
|
+ detector._capture_frame.return_value = None
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ http_client.post.assert_not_awaited()
|
|
|
+ detector._capture_frame.assert_awaited_once()
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert detector._states[1].next_check_at == 160
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("oversized", [False, True])
|
|
|
+ async def test_only_oversized_snapshots_are_prepared_before_upload(self, detector, http_client, clock, oversized):
|
|
|
+ frame = b"x" * (MAX_IMAGE_BYTES + int(oversized))
|
|
|
+ detector._capture_frame.return_value = frame
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ with patch(f"{MODULE}.prepare_frame", return_value=FAKE_JPEG) as prepare:
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ if oversized:
|
|
|
+ prepare.assert_called_once_with(frame)
|
|
|
+ else:
|
|
|
+ prepare.assert_not_called()
|
|
|
+ uploaded = http_client.post.await_args.kwargs["files"]["image"]
|
|
|
+ assert uploaded == ("snapshot.jpg", FAKE_JPEG if oversized else frame, "image/jpeg")
|
|
|
+ assert len(uploaded[1]) <= MAX_IMAGE_BYTES
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+
|
|
|
+ async def test_unprocessable_oversized_snapshot_is_not_uploaded(self, detector, http_client, clock):
|
|
|
+ frame = b"x" * (MAX_IMAGE_BYTES + 1)
|
|
|
+ detector._capture_frame.return_value = frame
|
|
|
+ with patch(f"{MODULE}.prepare_frame", side_effect=ValueError("private-image-details")) as prepare:
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action="pause_and_off"))
|
|
|
+ prepare.assert_called_once_with(frame)
|
|
|
+ http_client.post.assert_not_awaited()
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert "6 MiB" in detector.get_status()["last_error"]
|
|
|
+ assert "private-image-details" not in str(detector.get_status())
|
|
|
+ assert detector._states[1].next_check_at == 160
|
|
|
+
|
|
|
+ async def test_reset_during_capture_does_not_create_or_upload_context(self, detector, http_client, clock):
|
|
|
+ async def capture(printer_id):
|
|
|
+ detector.reset_printer(printer_id)
|
|
|
+ return FAKE_JPEG
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ http_client.post.assert_not_awaited()
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+
|
|
|
+ async def test_completion_during_processing_cannot_pause_next_print(self, detector, http_client, clock):
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ detector.reset_printer(1)
|
|
|
+ return httpx.Response(200, json=process_payload(PauseSuggested=True))
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action="pause"))
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "changes",
|
|
|
+ [
|
|
|
+ {"enabled": False},
|
|
|
+ {"api_key": "replacement-key"},
|
|
|
+ {"enabled_printers": set()},
|
|
|
+ {"action": "notify"},
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_saved_settings_invalidate_in_flight_actions(self, detector, http_client, manager, clock, changes):
|
|
|
+ detector._load_settings = AsyncMock(return_value=settings(**changes))
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ await detector.refresh_settings()
|
|
|
+ return httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(PauseSuggested=True, NextProcessIntervalSec={"Minimum": 120, "Recommended": 20}),
|
|
|
+ )
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await poll_once(detector, settings(action="pause"))
|
|
|
+
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ if "action" in changes:
|
|
|
+ # Action changes keep temporal context and server pacing, while
|
|
|
+ # discarding the decision made under the previous configuration.
|
|
|
+ assert detector._states[1].context is not None
|
|
|
+ assert detector._states[1].next_check_at == 220
|
|
|
+ elif "api_key" in changes:
|
|
|
+ assert detector._states[1].context is None
|
|
|
+ assert detector._states[1].next_check_at == 220
|
|
|
+ else:
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+
|
|
|
+ async def test_unknown_until_first_inference_returns(self, detector, http_client, clock):
|
|
|
+ async def capture(printer_id):
|
|
|
+ assert detector.get_per_printer()[printer_id] == {
|
|
|
+ "class": "unknown",
|
|
|
+ "print_quality": None,
|
|
|
+ "frame_count": 0,
|
|
|
+ "error": None,
|
|
|
+ "error_code": None,
|
|
|
+ }
|
|
|
+ return FAKE_JPEG
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+
|
|
|
+ async def test_invalid_json_is_not_a_safe_frame(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, text="not json: secret-api-key"),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert "secret-api-key" not in detector.get_status()["last_error"]
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("flag", ["WarningSuggested", "PauseSuggested"])
|
|
|
+ async def test_expired_context_is_recreated_without_repeating_actions(self, detector, http_client, clock, flag):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(**{flag: True})),
|
|
|
+ httpx.Response(
|
|
|
+ 500, json={"ErrorType": "OE_INTERNAL_ERROR", "ErrorDetails": "Failed to get context object."}
|
|
|
+ ),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=process_payload(**{flag: True})),
|
|
|
+ ]
|
|
|
+ for now in (100, 140, 200):
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].context.context_id == "new-context"
|
|
|
+ detector._dispatch_action.assert_awaited_once()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "status_code, payload",
|
|
|
+ [
|
|
|
+ (500, {"ErrorType": "OE_INTERNAL_ERROR", "ErrorDetails": "ML Model Exception"}),
|
|
|
+ (500, {"ErrorType": "OE_BACKEND_THROTTLED", "ErrorDetails": "Failed to get context object."}),
|
|
|
+ (404, {"ErrorType": "OE_INTERNAL_ERROR", "ErrorDetails": "Failed to get context object."}),
|
|
|
+ (500, {"ErrorType": "OE_INTERNAL_ERROR", "ErrorDetails": "Failed to get context object. secret-api-key"}),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_other_errors_retain_context_and_do_not_echo_details(
|
|
|
+ self, detector, http_client, clock, status_code, payload
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(status_code, json=payload),
|
|
|
+ httpx.Response(status_code, json=payload),
|
|
|
+ ]
|
|
|
+ for now in (100, 160):
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].context.context_id == "context-one"
|
|
|
+ assert detector._states[1].context.process_url == (FALLBACK_URL if status_code >= 500 else PRIMARY_URL)
|
|
|
+ assert "secret-api-key" not in detector.get_status()["last_error"]
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+
|
|
|
+ async def test_one_printers_success_does_not_hide_another_printers_error(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(503),
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ await detector._check_printer(2, printer_status(), settings())
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert detector.get_per_printer()[2]["class"] == "safe"
|
|
|
+ assert detector.get_status()["last_error"] is not None
|
|
|
+
|
|
|
+
|
|
|
+class TestFasterInspectionTiming:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "minimum, recommended, configured, delay",
|
|
|
+ [(5, 8, 20, 8), (5, 45, 5, 5), (5, 45, 20, 20), (40, 2, 30, 40)],
|
|
|
+ )
|
|
|
+ async def test_timing_hint_only_shortens_configured_interval_and_respects_minimum(
|
|
|
+ self, detector, http_client, clock, caplog, minimum, recommended, configured, delay
|
|
|
+ ):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": minimum, "Recommended": recommended},
|
|
|
+ FasterInspectionSuggested=True,
|
|
|
+ PrintQuality=2,
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=configured, action="pause_and_off"))
|
|
|
+ assert detector._states[1].recommended_interval == recommended
|
|
|
+ assert detector._states[1].next_check_at == 100 + delay
|
|
|
+ clock.return_value = 100 + delay - 0.1
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=configured, action="pause_and_off"))
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ results = [
|
|
|
+ record.message
|
|
|
+ for record in caplog.records
|
|
|
+ if record.name == MODULE and "inspection result" in record.message
|
|
|
+ ]
|
|
|
+ assert len(results) == 1
|
|
|
+ assert "faster_inspection_suggested=True" in results[0]
|
|
|
+ assert f"next_check_in={delay}s" in results[0]
|
|
|
+
|
|
|
+ async def test_latest_hint_changes_cadence_and_false_or_missing_restores_setting(
|
|
|
+ self, detector, http_client, clock
|
|
|
+ ):
|
|
|
+ responses = [(True, 8), (True, 6), (False, 5), (True, 5), (None, 5)]
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload())] + [
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": recommended},
|
|
|
+ **({"FasterInspectionSuggested": hint} if hint is not None else {}),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ for hint, recommended in responses
|
|
|
+ ]
|
|
|
+ context = None
|
|
|
+ for now, next_check, active_interval in [
|
|
|
+ (100, 108, 8),
|
|
|
+ (108, 114, 6),
|
|
|
+ (114, 134, None),
|
|
|
+ (134, 139, 5),
|
|
|
+ (139, 159, None),
|
|
|
+ ]:
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ if context is None:
|
|
|
+ context = state.context
|
|
|
+ assert state.context is context
|
|
|
+ assert state.recommended_interval == active_interval
|
|
|
+ assert state.next_check_at == next_check
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [CREATE_CONTEXT_URL] + [PRIMARY_URL] * 5
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_interval_edit_preserves_hint_then_uses_new_setting_when_hint_clears(
|
|
|
+ self, detector, http_client, manager, clock
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=False
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ context = detector._states[1].context
|
|
|
+ clock.return_value = 101
|
|
|
+ await poll_once(detector, settings(poll_interval=30))
|
|
|
+ assert detector._states[1].next_check_at == 105
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ clock.return_value = 105
|
|
|
+ await poll_once(detector, settings(poll_interval=30))
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert detector._states[1].recommended_interval is None
|
|
|
+ assert detector._states[1].next_check_at == 135
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "hint, interval_fields",
|
|
|
+ [
|
|
|
+ (None, {"Recommended": 5}),
|
|
|
+ (1, {"Recommended": 5}),
|
|
|
+ ("true", {"Recommended": 5}),
|
|
|
+ (True, {}),
|
|
|
+ (True, {"Recommended": None}),
|
|
|
+ (True, {"Recommended": True}),
|
|
|
+ (True, {"Recommended": 0}),
|
|
|
+ (True, {"Recommended": -1}),
|
|
|
+ (True, {"Recommended": 2.5}),
|
|
|
+ (True, {"Recommended": "private-recommendation"}),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_invalid_optional_hint_restores_setting_without_losing_warning(
|
|
|
+ self, detector, http_client, clock, caplog, hint, interval_fields
|
|
|
+ ):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, **interval_fields},
|
|
|
+ FasterInspectionSuggested=hint,
|
|
|
+ WarningSuggested=True,
|
|
|
+ PrintQuality=4,
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ clock.return_value = 105
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ assert state.recommended_interval is None
|
|
|
+ assert state.next_check_at == 125
|
|
|
+ assert state.frame_count == 2
|
|
|
+ assert state.error is None
|
|
|
+ assert state.verdict == "warning"
|
|
|
+ detector._dispatch_action.assert_awaited_once_with(1, "notify", "test-print", 4)
|
|
|
+ assert "private-recommendation" not in "\n".join(
|
|
|
+ record.message for record in caplog.records if record.name == MODULE
|
|
|
+ )
|
|
|
+
|
|
|
+ async def test_timing_hint_is_scoped_to_printer_and_print(self, detector, http_client, manager, clock):
|
|
|
+ manager.get_all_statuses.return_value[2] = printer_status(subtask_id="other-print")
|
|
|
+ fast = process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True)
|
|
|
+ normal = process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 5})
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=fast),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="other-context")),
|
|
|
+ httpx.Response(200, json=normal),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=normal),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1].next_check_at == 105
|
|
|
+ assert detector._states[2].next_check_at == 120
|
|
|
+ old_state = detector._states[1]
|
|
|
+ manager.get_all_statuses.return_value[1].subtask_id = "new-print"
|
|
|
+ clock.return_value = 101
|
|
|
+
|
|
|
+ async def capture(printer_id):
|
|
|
+ assert printer_id == 1
|
|
|
+ assert detector._states[printer_id].recommended_interval is None
|
|
|
+ assert detector._states[printer_id].next_check_at == 121
|
|
|
+ return FAKE_JPEG
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is not old_state
|
|
|
+ assert detector._states[1].next_check_at == 121
|
|
|
+ assert detector._states[2].next_check_at == 120
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("changes", [{"api_key": "new-key"}, {"confidence": "highest"}])
|
|
|
+ async def test_context_configuration_change_clears_timing_hint(
|
|
|
+ self, detector, http_client, manager, clock, changes
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ clock.return_value = 101
|
|
|
+ await poll_once(detector, settings(**changes))
|
|
|
+ assert detector._states[1].recommended_interval is None
|
|
|
+ assert detector._states[1].context is None
|
|
|
+ assert detector._states[1].next_check_at == 120
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+
|
|
|
+ async def test_expired_context_clears_timing_hint_and_respects_backoff(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 500, json={"ErrorType": "OE_INTERNAL_ERROR", "ErrorDetails": "Failed to get context object."}
|
|
|
+ ),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 5, "Recommended": 5})),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ clock.return_value = 105
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ assert state.context is None
|
|
|
+ assert state.recommended_interval is None
|
|
|
+ assert state.next_check_at == 165
|
|
|
+ clock.return_value = 164
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ clock.return_value = 165
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert state.context.context_id == "new-context"
|
|
|
+ assert state.next_check_at == 185
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "response, delay, host",
|
|
|
+ [
|
|
|
+ (httpx.ConnectError("private-transport-details"), 60, FALLBACK_URL),
|
|
|
+ (
|
|
|
+ httpx.Response(429, headers={"Retry-After": "300"}, json={"ErrorType": "OE_CONTEXT_RATE_LIMITED"}),
|
|
|
+ 300,
|
|
|
+ PRIMARY_URL,
|
|
|
+ ),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_active_timing_hint_never_shortens_error_backoff(
|
|
|
+ self, detector, http_client, clock, response, delay, host
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ response,
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ clock.return_value = 105
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ assert state.next_check_at == 105 + delay
|
|
|
+ assert state.context.process_url == host
|
|
|
+ clock.return_value = 105 + delay - 1
|
|
|
+ await detector._check_printer(1, printer_status(), settings(poll_interval=5))
|
|
|
+ assert state.next_check_at == 105 + delay
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_invalid_assessment_keeps_latest_minimum_without_accepting_new_hint(
|
|
|
+ self, detector, http_client, clock
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 90, "Recommended": 2},
|
|
|
+ FasterInspectionSuggested=True,
|
|
|
+ PrintQuality=None,
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ clock.return_value = 105
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ assert state.interval == 90
|
|
|
+ assert state.recommended_interval == 5
|
|
|
+ assert state.next_check_at == 195
|
|
|
+ assert state.frame_count == 1
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_canceled_fast_inspection_preserves_minimum_before_retry(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 10, "Recommended": 5}, FasterInspectionSuggested=True
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+
|
|
|
+ async def cancel_response(url, **kwargs):
|
|
|
+ clock.return_value = 115
|
|
|
+ raise asyncio.CancelledError
|
|
|
+
|
|
|
+ http_client.post.side_effect = cancel_response
|
|
|
+ clock.return_value = 110
|
|
|
+ with pytest.raises(asyncio.CancelledError):
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].recommended_interval == 5
|
|
|
+ assert detector._states[1].next_check_at == 125
|
|
|
+ clock.return_value = 124
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+
|
|
|
+ async def test_stale_response_does_not_apply_timing_hint(self, detector, http_client, clock):
|
|
|
+ status = printer_status()
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ status.subtask_id = "new-print"
|
|
|
+ return httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5},
|
|
|
+ FasterInspectionSuggested=True,
|
|
|
+ PauseSuggested=True,
|
|
|
+ ),
|
|
|
+ )
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, status, settings(action="pause"))
|
|
|
+ assert detector._states[1].recommended_interval is None
|
|
|
+ assert detector._states[1].frame_count == 0
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+
|
|
|
+class TestDebugLogging:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "warning, pause, quality, verdict, minimum, interval",
|
|
|
+ [(False, False, 9, "safe", 5, 20), (True, False, 4, "warning", 40, 40), (True, True, 1, "failure", 5, 20)],
|
|
|
+ )
|
|
|
+ async def test_each_due_inspection_logs_results_and_context_reuse(
|
|
|
+ self, detector, http_client, clock, caplog, warning, pause, quality, verdict, minimum, interval
|
|
|
+ ):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+ payload = process_payload(
|
|
|
+ WarningSuggested=warning,
|
|
|
+ PauseSuggested=pause,
|
|
|
+ PrintQuality=quality,
|
|
|
+ NextProcessIntervalSec={"Minimum": minimum, "Recommended": 20},
|
|
|
+ ExtraDetails="private-response-details",
|
|
|
+ )
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=payload),
|
|
|
+ httpx.Response(200, json=payload),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ first_messages = [record.message for record in caplog.records if record.name == MODULE]
|
|
|
+ clock.return_value = 100 + interval - 1
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert [record.message for record in caplog.records if record.name == MODULE] == first_messages
|
|
|
+
|
|
|
+ clock.return_value = 100 + interval
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ records = [record for record in caplog.records if record.name == MODULE]
|
|
|
+ messages = [record.message for record in records]
|
|
|
+ assert all(record.levelno == logging.DEBUG for record in records)
|
|
|
+ assert messages.count("OctoEverywhere context created for printer 1") == 1
|
|
|
+ for frame, reused in ((1, False), (2, True)):
|
|
|
+ assert (
|
|
|
+ f"OctoEverywhere inspection starting for printer 1 (frame={frame}, context_reused={reused})"
|
|
|
+ ) in messages
|
|
|
+ assert (
|
|
|
+ f"OctoEverywhere inspection result for printer 1: frame={frame}, verdict={verdict}, "
|
|
|
+ f"print_quality={quality}, warning_suggested={warning}, pause_suggested={pause}, "
|
|
|
+ "faster_inspection_suggested=False, "
|
|
|
+ f"minimum_interval={minimum}s, next_check_in={interval}s"
|
|
|
+ ) in messages
|
|
|
+ assert len(messages) == 5
|
|
|
+ log_text = "\n".join(messages)
|
|
|
+ for private_value in ("secret-api-key", "context-one", PRIMARY_URL, FALLBACK_URL, "private-response-details"):
|
|
|
+ assert private_value not in log_text
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+
|
|
|
+ async def test_repeated_errors_log_each_retry_with_safe_error_code(self, detector, http_client, clock, caplog):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+ response = httpx.Response(
|
|
|
+ 429,
|
|
|
+ headers={"Retry-After": "30"},
|
|
|
+ json={"ErrorType": "OE_CONTEXT_RATE_LIMITED", "ErrorDetails": "private-response-details secret-api-key"},
|
|
|
+ )
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), response, response]
|
|
|
+ for now in (100, 160):
|
|
|
+ clock.return_value = now
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+
|
|
|
+ records = [record for record in caplog.records if record.name == MODULE]
|
|
|
+ failures = [record for record in records if "inspection failed" in record.message]
|
|
|
+ assert len(failures) == 2
|
|
|
+ assert all(record.levelno == logging.DEBUG for record in failures)
|
|
|
+ for record, delay in zip(failures, (60, 120), strict=True):
|
|
|
+ assert f"error_code=OE_CONTEXT_RATE_LIMITED, retry_in={delay:.1f}s" in record.message
|
|
|
+ assert API_ERRORS["OE_CONTEXT_RATE_LIMITED"] in record.message
|
|
|
+ assert len([record for record in records if record.levelno == logging.WARNING]) == 1
|
|
|
+ assert not any("inspection result" in record.message for record in records)
|
|
|
+ log_text = "\n".join(record.message for record in records)
|
|
|
+ for private_value in ("secret-api-key", "context-one", PRIMARY_URL, FALLBACK_URL, "private-response-details"):
|
|
|
+ assert private_value not in log_text
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "response, expected_error",
|
|
|
+ [
|
|
|
+ (
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ PrintQuality="private-invalid-quality",
|
|
|
+ WarningSuggested="private-invalid-warning",
|
|
|
+ PauseSuggested="private-invalid-pause",
|
|
|
+ ),
|
|
|
+ ),
|
|
|
+ "invalid print assessment",
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(NextProcessIntervalSec={"Minimum": "private-invalid-interval"}),
|
|
|
+ ),
|
|
|
+ "invalid processing interval",
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ httpx.Response(
|
|
|
+ 500,
|
|
|
+ json={"ErrorType": "private-unknown-code", "ErrorDetails": "private-error-details secret-api-key"},
|
|
|
+ ),
|
|
|
+ "API request failed (HTTP 500)",
|
|
|
+ ),
|
|
|
+ (httpx.ConnectError(f"private-transport-details {PRIMARY_URL} secret-api-key"), "Could not reach"),
|
|
|
+ (httpx.Response(200, text="private-raw-payload secret-api-key"), "invalid response"),
|
|
|
+ ],
|
|
|
+ ids=["invalid-assessment", "invalid-interval", "unknown-error", "transport-error", "invalid-json"],
|
|
|
+ )
|
|
|
+ async def test_failures_do_not_log_remote_payloads_or_credentials(
|
|
|
+ self, detector, http_client, clock, caplog, response, expected_error
|
|
|
+ ):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), response]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ records = [record for record in caplog.records if record.name == MODULE]
|
|
|
+ failures = [record for record in records if "inspection failed" in record.message]
|
|
|
+ assert len(failures) == 1
|
|
|
+ assert failures[0].levelno == logging.DEBUG
|
|
|
+ assert expected_error in failures[0].message
|
|
|
+ assert "error_code=None, retry_in=60.0s" in failures[0].message
|
|
|
+ assert not any("inspection result" in record.message for record in records)
|
|
|
+ log_text = "\n".join(record.message for record in records)
|
|
|
+ for private_value in ("secret-api-key", "context-one", PRIMARY_URL, FALLBACK_URL, "private-"):
|
|
|
+ assert private_value not in log_text
|
|
|
+
|
|
|
+ async def test_stale_result_logs_discard_without_logging_assessment(self, detector, http_client, clock, caplog):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ detector.reset_printer(1)
|
|
|
+ return httpx.Response(200, json=process_payload(PauseSuggested=True))
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, printer_status(), settings(action="pause"))
|
|
|
+ records = [record for record in caplog.records if record.name == MODULE]
|
|
|
+ discarded = [record for record in records if "inspection discarded" in record.message]
|
|
|
+ assert len(discarded) == 1
|
|
|
+ assert discarded[0].levelno == logging.DEBUG
|
|
|
+ assert not any("inspection result" in record.message for record in records)
|
|
|
+ assert not any("inspection failed" in record.message for record in records)
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_canceled_inspection_logs_cancellation_without_exception_details(
|
|
|
+ self, detector, http_client, clock, caplog
|
|
|
+ ):
|
|
|
+ caplog.set_level(logging.DEBUG, logger=MODULE)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ asyncio.CancelledError("private-cancellation-details secret-api-key"),
|
|
|
+ ]
|
|
|
+ with pytest.raises(asyncio.CancelledError):
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ records = [record for record in caplog.records if record.name == MODULE]
|
|
|
+ canceled = [record for record in records if "inspection canceled" in record.message]
|
|
|
+ assert len(canceled) == 1
|
|
|
+ assert canceled[0].levelno == logging.DEBUG
|
|
|
+ assert not any("inspection result" in record.message for record in records)
|
|
|
+ assert not any("inspection failed" in record.message for record in records)
|
|
|
+ log_text = "\n".join(record.message for record in records)
|
|
|
+ assert "private-cancellation-details" not in log_text
|
|
|
+ assert "secret-api-key" not in log_text
|
|
|
+
|
|
|
+
|
|
|
+class TestErrorCodes:
|
|
|
+ async def test_connection_exposes_quota_code_and_safe_message(self, detector, http_client):
|
|
|
+ http_client.post.return_value = quota_response()
|
|
|
+ result = await detector.test_connection("secret-api-key")
|
|
|
+ assert result == {
|
|
|
+ "ok": False,
|
|
|
+ "status_code": 429,
|
|
|
+ "error": API_ERRORS[USAGE_LIMIT_ERROR],
|
|
|
+ "error_code": USAGE_LIMIT_ERROR,
|
|
|
+ }
|
|
|
+ assert result["error"] == "API usage limit reached. Set up billing to continue."
|
|
|
+ assert "secret-api-key" not in result["error"]
|
|
|
+ detector._capture_frame.assert_not_awaited()
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "unknown_code", [None, "OE_FUTURE_ERROR_secret-api-key", ["secret-api-key"], {"secret": "secret-api-key"}]
|
|
|
+ )
|
|
|
+ async def test_unknown_remote_codes_are_not_exposed(self, detector, http_client, clock, unknown_code):
|
|
|
+ response = httpx.Response(429, json={"ErrorType": unknown_code, "ErrorDetails": "secret-api-key"})
|
|
|
+ http_client.post.return_value = response
|
|
|
+ result = await detector.test_connection("secret-api-key")
|
|
|
+ assert result["error_code"] is None
|
|
|
+ assert "secret-api-key" not in result["error"]
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), response]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ status = detector.get_status()
|
|
|
+ assert status["last_error_code"] is None
|
|
|
+ assert status["per_printer"][1]["error_code"] is None
|
|
|
+ assert "secret-api-key" not in str(status)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "error_code",
|
|
|
+ [
|
|
|
+ "OE_INVALID_API_KEY",
|
|
|
+ "OE_API_KEY_DISABLED",
|
|
|
+ "OE_API_KEY_BLOCKED_PAYMENT_FAILED",
|
|
|
+ IP_RESTRICTED_ERROR,
|
|
|
+ USAGE_LIMIT_ERROR,
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ @pytest.mark.parametrize("stage", ["context", "inspection"])
|
|
|
+ async def test_account_errors_stop_all_printers_without_automatic_retries(
|
|
|
+ self, detector, http_client, manager, clock, error_code, stage
|
|
|
+ ):
|
|
|
+ manager.get_all_statuses.return_value[2] = printer_status(subtask_id="print-2")
|
|
|
+ response = httpx.Response(
|
|
|
+ 429 if error_code == USAGE_LIMIT_ERROR else 403,
|
|
|
+ headers={"Retry-After": "30"},
|
|
|
+ json={"ErrorType": error_code, "ErrorDetails": "secret-api-key"},
|
|
|
+ )
|
|
|
+ http_client.post.side_effect = (
|
|
|
+ [httpx.Response(200, json=context_payload()), response] if stage == "inspection" else [response]
|
|
|
+ )
|
|
|
+ await poll_once(detector, settings(action="pause_and_off"))
|
|
|
+ calls = http_client.post.await_count
|
|
|
+ detector._capture_frame.assert_awaited_once_with(1)
|
|
|
+ for now in (130, 3700, 100 + 31 * 24 * 3600):
|
|
|
+ clock.return_value = now
|
|
|
+ await poll_once(detector, settings(poll_interval=5, action="pause_and_off"))
|
|
|
+ assert http_client.post.await_count == calls
|
|
|
+ detector._capture_frame.assert_awaited_once_with(1)
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ status = detector.get_status()
|
|
|
+ assert status["last_error_code"] == error_code
|
|
|
+ assert status["last_error"] == API_ERRORS[error_code]
|
|
|
+ assert set(status["per_printer"]) == {1, 2}
|
|
|
+ assert all(entry["error_code"] == error_code for entry in status["per_printer"].values())
|
|
|
+ assert all(entry["class"] == "error" for entry in status["per_printer"].values())
|
|
|
+ assert status["history"] == []
|
|
|
+ assert "secret-api-key" not in str(status)
|
|
|
+ if stage == "inspection":
|
|
|
+ assert detector._states[1].context.process_url == PRIMARY_URL
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("status_code", [401, 403])
|
|
|
+ async def test_unrecognized_auth_response_still_blocks_the_account(self, detector, http_client, clock, status_code):
|
|
|
+ http_client.post.return_value = httpx.Response(status_code, text="secret-api-key")
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ clock.return_value = 1000
|
|
|
+ await detector._check_printer(2, printer_status(), settings())
|
|
|
+ assert detector.get_status()["last_error_code"] == "OE_INVALID_API_KEY"
|
|
|
+ http_client.post.assert_awaited_once()
|
|
|
+ detector._capture_frame.assert_awaited_once()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "change", ["confidence", "interval", "action", "selection", "disabled", "reset", "new_print", "disconnect"]
|
|
|
+ )
|
|
|
+ async def test_ordinary_settings_and_lifecycle_changes_do_not_clear_account_block(
|
|
|
+ self, detector, http_client, manager, clock, change
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), quota_response()]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ clock.return_value = 1000
|
|
|
+ configuration = settings()
|
|
|
+ if change == "reset":
|
|
|
+ detector.reset_printer(1)
|
|
|
+ elif change == "new_print":
|
|
|
+ manager.get_all_statuses.return_value[1] = printer_status(subtask_id="print-2")
|
|
|
+ elif change == "disconnect":
|
|
|
+ manager.is_connected.return_value = False
|
|
|
+ else:
|
|
|
+ configuration.update(
|
|
|
+ {
|
|
|
+ "confidence": {"confidence": "high"},
|
|
|
+ "interval": {"poll_interval": 5},
|
|
|
+ "action": {"action": "pause_and_off"},
|
|
|
+ "selection": {"enabled_printers": set()},
|
|
|
+ "disabled": {"enabled": False},
|
|
|
+ }[change]
|
|
|
+ )
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ manager.is_connected.return_value = True
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ status = detector.get_status()
|
|
|
+ assert status["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert status["last_error"] == API_ERRORS[USAGE_LIMIT_ERROR]
|
|
|
+ assert status["per_printer"][1]["error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ detector._capture_frame.assert_awaited_once()
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("tested_key, recovered", [("secret-api-key", True), ("another-key", False)])
|
|
|
+ async def test_only_successful_test_of_configured_key_resumes_checks(
|
|
|
+ self, detector, http_client, manager, clock, tested_key, recovered
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ ip_restricted_response(),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="test-context")),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ context = detector._states[1].context
|
|
|
+ assert (await detector.test_connection(tested_key))["ok"] is True
|
|
|
+ clock.return_value = 120
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert http_client.post.await_count == (4 if recovered else 3)
|
|
|
+ assert detector.get_per_printer()[1]["class"] == ("safe" if recovered else "error")
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert detector._states[1].context.process_url == PRIMARY_URL
|
|
|
+
|
|
|
+ async def test_recovery_preserves_context_actions_and_server_deadline(self, detector, http_client, manager, clock):
|
|
|
+ assessment = process_payload(NextProcessIntervalSec={"Minimum": 300, "Recommended": 20}, PauseSuggested=True)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=assessment),
|
|
|
+ quota_response(),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="test-context")),
|
|
|
+ httpx.Response(200, json=assessment),
|
|
|
+ ]
|
|
|
+ configuration = settings(action="pause_and_off")
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+ clock.return_value = 400
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert (await detector.test_connection("secret-api-key"))["ok"] is True
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.warning_fired and state.action_fired
|
|
|
+ assert state.interval == 300
|
|
|
+ assert state.next_check_at == 700
|
|
|
+ clock.return_value = 699
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert http_client.post.await_count == 4
|
|
|
+ clock.return_value = 700
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert http_client.post.await_count == 5
|
|
|
+ assert http_client.post.await_args.args == (PRIMARY_URL,)
|
|
|
+ assert detector.get_status()["last_error_code"] is None
|
|
|
+ assert detector.get_status()["last_error"] is None
|
|
|
+ assert state.context is context
|
|
|
+ detector._dispatch_action.assert_awaited_once()
|
|
|
+
|
|
|
+ async def test_saving_replacement_key_recovers_without_repeating_actions(
|
|
|
+ self, detector, http_client, manager, clock
|
|
|
+ ):
|
|
|
+ assessment = process_payload(NextProcessIntervalSec={"Minimum": 300, "Recommended": 20}, PauseSuggested=True)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=assessment),
|
|
|
+ quota_response(),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=assessment),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings(action="pause"))
|
|
|
+ state = detector._states[1]
|
|
|
+ clock.return_value = 400
|
|
|
+ await poll_once(detector, settings(action="pause"))
|
|
|
+ configuration = settings(api_key="replacement-key", action="pause")
|
|
|
+ clock.return_value = 401
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert detector.get_status()["last_error_code"] is None
|
|
|
+ assert state.context is None
|
|
|
+ assert state.interval == 300
|
|
|
+ assert state.next_check_at == 700
|
|
|
+ assert state.warning_fired and state.action_fired
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ clock.return_value = 700
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert state.context.context_id == "new-context"
|
|
|
+ creation = http_client.post.await_args_list[3]
|
|
|
+ assert creation.args == (CREATE_CONTEXT_URL,)
|
|
|
+ assert creation.kwargs["headers"]["X-API-Key"] == "replacement-key"
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "failure"
|
|
|
+ detector._dispatch_action.assert_awaited_once()
|
|
|
+
|
|
|
+ async def test_successful_test_cannot_confirm_remaining_inspection_allowance(
|
|
|
+ self, detector, http_client, manager, clock
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ quota_response(),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="test-context")),
|
|
|
+ quota_response(),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert (await detector.test_connection("secret-api-key"))["ok"] is True
|
|
|
+ clock.return_value = 120
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ clock.return_value = 1000
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert http_client.post.await_count == 4
|
|
|
+ assert detector._capture_frame.await_count == 2
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("response", [httpx.Response(503), httpx.ReadTimeout("private-details")])
|
|
|
+ async def test_failed_test_does_not_clear_account_block(self, detector, http_client, manager, clock, response):
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), quota_response(), response]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert (await detector.test_connection("secret-api-key"))["ok"] is False
|
|
|
+ clock.return_value = 1000
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+ detector._capture_frame.assert_awaited_once()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("tested_key, blocks", [("secret-api-key", True), ("another-key", False)])
|
|
|
+ async def test_account_error_testing_configured_key_stops_monitoring(
|
|
|
+ self, detector, http_client, manager, clock, tested_key, blocks
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ip_restricted_response(),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert (await detector.test_connection(tested_key))["ok"] is False
|
|
|
+ clock.return_value = 140
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert http_client.post.await_count == (3 if blocks else 4)
|
|
|
+ assert detector.get_status()["last_error_code"] == (IP_RESTRICTED_ERROR if blocks else None)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("replace_key", [False, True])
|
|
|
+ async def test_stale_successful_test_cannot_clear_a_newer_account_block(
|
|
|
+ self, detector, http_client, manager, clock, replace_key
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ if kwargs["headers"]["X-API-Key"] == "secret-api-key":
|
|
|
+ started.set()
|
|
|
+ await release.wait()
|
|
|
+ return httpx.Response(200, json=context_payload(ContextId="test-context"))
|
|
|
+ return quota_response()
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ test = asyncio.create_task(detector.test_connection("secret-api-key"))
|
|
|
+ try:
|
|
|
+ await asyncio.wait_for(started.wait(), timeout=1)
|
|
|
+ configuration = settings(api_key="replacement-key") if replace_key else settings()
|
|
|
+ clock.return_value = 140
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ release.set()
|
|
|
+ assert (await asyncio.wait_for(test, timeout=1))["ok"] is True
|
|
|
+ clock.return_value = 1000
|
|
|
+ captures = detector._capture_frame.await_count
|
|
|
+ await poll_once(detector, configuration)
|
|
|
+ assert detector._capture_frame.await_count == captures
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ finally:
|
|
|
+ release.set()
|
|
|
+ await asyncio.gather(test, return_exceptions=True)
|
|
|
+
|
|
|
+ async def test_stale_failed_test_cannot_block_a_replacement_key(self, detector, http_client, manager, clock):
|
|
|
+ detector._apply_settings(settings())
|
|
|
+ started = asyncio.Event()
|
|
|
+ release = asyncio.Event()
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ started.set()
|
|
|
+ await release.wait()
|
|
|
+ return ip_restricted_response()
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ test = asyncio.create_task(detector.test_connection("secret-api-key"))
|
|
|
+ try:
|
|
|
+ await asyncio.wait_for(started.wait(), timeout=1)
|
|
|
+ detector._apply_settings(settings(api_key="replacement-key"))
|
|
|
+ release.set()
|
|
|
+ assert (await asyncio.wait_for(test, timeout=1))["ok"] is False
|
|
|
+ assert detector.get_status()["last_error_code"] is None
|
|
|
+ finally:
|
|
|
+ release.set()
|
|
|
+ await asyncio.gather(test, return_exceptions=True)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("change", ["paused", "finished", "reset"])
|
|
|
+ async def test_account_error_from_ending_print_still_blocks_other_printers(
|
|
|
+ self, detector, http_client, clock, change
|
|
|
+ ):
|
|
|
+ status = printer_status()
|
|
|
+ detector._apply_settings(settings())
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ if change == "reset":
|
|
|
+ detector.reset_printer(1)
|
|
|
+ else:
|
|
|
+ status.state = "PAUSE" if change == "paused" else "FINISH"
|
|
|
+ return quota_response()
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, status, settings())
|
|
|
+ await detector._check_printer(2, printer_status(subtask_id="print-2"), settings())
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert detector.get_per_printer()[2]["error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ detector._capture_frame.assert_awaited_once_with(1)
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("changes", [{"api_key": "replacement-key"}, {"confidence": "high"}])
|
|
|
+ async def test_account_error_from_old_configuration_cannot_block_new_configuration(
|
|
|
+ self, detector, http_client, clock, changes
|
|
|
+ ):
|
|
|
+ detector._apply_settings(settings())
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ detector._apply_settings(settings(**changes))
|
|
|
+ return quota_response()
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector.get_status()["last_error_code"] is None
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ clock.return_value = 120
|
|
|
+ await detector._check_printer(1, printer_status(), settings(**changes))
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+ assert http_client.post.await_count == 4
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("stage", ["camera", "inspection"])
|
|
|
+ @pytest.mark.parametrize("ignore_cancellation", [False, True])
|
|
|
+ async def test_account_block_cancels_sibling_checks_and_discards_late_decisions(
|
|
|
+ self, detector, http_client, manager, clock, stage, ignore_cancellation
|
|
|
+ ):
|
|
|
+ manager.get_all_statuses.return_value = {2: printer_status(subtask_id="print-2"), 1: printer_status()}
|
|
|
+ started = asyncio.Event()
|
|
|
+ canceled = asyncio.Event()
|
|
|
+
|
|
|
+ async def wait_for_block():
|
|
|
+ started.set()
|
|
|
+ try:
|
|
|
+ await asyncio.Event().wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ canceled.set()
|
|
|
+ if not ignore_cancellation:
|
|
|
+ raise
|
|
|
+
|
|
|
+ async def capture(printer_id):
|
|
|
+ if printer_id == 2 and stage == "camera":
|
|
|
+ await wait_for_block()
|
|
|
+ if printer_id == 1:
|
|
|
+ await started.wait()
|
|
|
+ return FAKE_JPEG + bytes([printer_id])
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ if kwargs["files"]["image"][1][-1] == 1:
|
|
|
+ return quota_response()
|
|
|
+ await wait_for_block()
|
|
|
+ return httpx.Response(200, json=process_payload(PauseSuggested=True))
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._poll_once(settings(action="pause_and_off"))
|
|
|
+ checks = list(detector._checks.values())
|
|
|
+ try:
|
|
|
+ await asyncio.wait_for(asyncio.gather(*checks, return_exceptions=True), timeout=1)
|
|
|
+ assert canceled.is_set()
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert detector._states[2].frame_count == 0
|
|
|
+ assert not detector._states[2].action_fired
|
|
|
+ assert detector.get_status()["history"] == []
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ captures = detector._capture_frame.await_count
|
|
|
+ calls = http_client.post.await_count
|
|
|
+ clock.return_value = 1000
|
|
|
+ await poll_once(detector, settings(action="pause_and_off"))
|
|
|
+ assert detector._capture_frame.await_count == captures
|
|
|
+ assert http_client.post.await_count == calls
|
|
|
+ finally:
|
|
|
+ detector.stop()
|
|
|
+ await asyncio.gather(*checks, return_exceptions=True)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "response, expected_url",
|
|
|
+ [
|
|
|
+ (httpx.Response(400, json={"ErrorType": "OE_BAD_ARGS"}), PRIMARY_URL),
|
|
|
+ (httpx.Response(400, json={"ErrorType": "OE_ARGS_PARSE_FAILED"}), PRIMARY_URL),
|
|
|
+ (httpx.Response(422, json={"ErrorType": "OE_IMAGE_DECODE_FAILED"}), PRIMARY_URL),
|
|
|
+ (httpx.Response(429, json={"ErrorType": "OE_CONTEXT_RATE_LIMITED"}), PRIMARY_URL),
|
|
|
+ (httpx.Response(500, json={"ErrorType": "OE_BAD_ARGS"}), PRIMARY_URL),
|
|
|
+ (httpx.Response(200, text="not-json"), PRIMARY_URL),
|
|
|
+ (httpx.Response(503, json={"ErrorType": "OE_BACKEND_THROTTLED"}), FALLBACK_URL),
|
|
|
+ (httpx.Response(500, json={"ErrorType": "OE_INTERNAL_ERROR"}), FALLBACK_URL),
|
|
|
+ (httpx.Response(502, text="not-json"), FALLBACK_URL),
|
|
|
+ (httpx.ConnectError("private-details"), FALLBACK_URL),
|
|
|
+ (httpx.ReadTimeout("private-details"), FALLBACK_URL),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_fallback_is_only_used_for_transport_or_server_failures(
|
|
|
+ self, detector, http_client, clock, response, expected_url
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ response,
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ context = detector._states[1].context
|
|
|
+ assert context.process_url == expected_url
|
|
|
+ clock.return_value = 160
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert http_client.post.await_args.args == (expected_url,)
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+
|
|
|
+ async def test_loop_error_does_not_hide_account_block(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), quota_response()]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ detector._load_settings = AsyncMock(side_effect=RuntimeError("secret-api-key"))
|
|
|
+ with (
|
|
|
+ patch(f"{MODULE}.asyncio.sleep", new=AsyncMock(side_effect=asyncio.CancelledError)),
|
|
|
+ pytest.raises(asyncio.CancelledError),
|
|
|
+ ):
|
|
|
+ await detector._loop()
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert detector.get_status()["last_error"] == API_ERRORS[USAGE_LIMIT_ERROR]
|
|
|
+
|
|
|
+ async def test_action_error_does_not_hide_account_block(self, detector, http_client, clock):
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), quota_response()]
|
|
|
+ await detector._check_printer(1, printer_status(), settings())
|
|
|
+ detector._dispatch_action = OctoEverywhereDetectionService._dispatch_action.__get__(detector)
|
|
|
+ with patch(
|
|
|
+ "backend.app.services.octoeverywhere_actions.execute_action", side_effect=RuntimeError("secret-api-key")
|
|
|
+ ):
|
|
|
+ await detector._dispatch_action(1, "notify", "test-print", 1)
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert detector.get_status()["last_error"] == API_ERRORS[USAGE_LIMIT_ERROR]
|
|
|
+
|
|
|
+
|
|
|
+class TestScheduling:
|
|
|
+ @pytest.mark.parametrize("slow_stage", ["camera", "api"])
|
|
|
+ @pytest.mark.parametrize("faster, interval", [(False, 20), (True, 5)])
|
|
|
+ async def test_slow_printer_does_not_delay_other_inspections(
|
|
|
+ self, detector, http_client, manager, clock, slow_stage, faster, interval
|
|
|
+ ):
|
|
|
+ blocked = asyncio.Event()
|
|
|
+ manager.get_all_statuses.return_value[2] = printer_status(subtask_id="print-2")
|
|
|
+
|
|
|
+ async def capture(printer_id):
|
|
|
+ if printer_id == 1 and slow_stage == "camera":
|
|
|
+ blocked.set()
|
|
|
+ await asyncio.Event().wait()
|
|
|
+ return FAKE_JPEG + bytes([printer_id])
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ if kwargs["files"]["image"][1][-1] == 1 and slow_stage == "api":
|
|
|
+ blocked.set()
|
|
|
+ await asyncio.Event().wait()
|
|
|
+ return httpx.Response(
|
|
|
+ 200,
|
|
|
+ json=process_payload(
|
|
|
+ NextProcessIntervalSec={"Minimum": 5, "Recommended": 5},
|
|
|
+ FasterInspectionSuggested=faster,
|
|
|
+ ),
|
|
|
+ )
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ poll = asyncio.create_task(detector._poll_once(settings()))
|
|
|
+ try:
|
|
|
+ await asyncio.wait_for(blocked.wait(), timeout=1)
|
|
|
+ await asyncio.wait_for(asyncio.shield(poll), timeout=1)
|
|
|
+ slow_check = detector._checks[1]
|
|
|
+ context = detector._states[2].context
|
|
|
+ assert detector._states[2].frame_count == 1
|
|
|
+
|
|
|
+ clock.return_value = 100 + interval - 1
|
|
|
+ await asyncio.wait_for(detector._poll_once(settings()), timeout=1)
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert detector._states[2].frame_count == 1
|
|
|
+
|
|
|
+ for frame_count in (2, 3):
|
|
|
+ clock.return_value = 100 + interval * (frame_count - 1)
|
|
|
+ await asyncio.wait_for(detector._poll_once(settings()), timeout=1)
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ assert detector._states[2].frame_count == frame_count
|
|
|
+ assert detector._states[2].context is context
|
|
|
+ assert detector._checks[1] is slow_check
|
|
|
+ assert not slow_check.done()
|
|
|
+
|
|
|
+ captures = [call.args[0] for call in detector._capture_frame.await_args_list]
|
|
|
+ assert captures.count(1) == 1
|
|
|
+ assert captures.count(2) == 3
|
|
|
+ finally:
|
|
|
+ checks = list(detector._checks.values())
|
|
|
+ detector.stop()
|
|
|
+ poll.cancel()
|
|
|
+ await asyncio.gather(poll, *checks, return_exceptions=True)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("change", ["paused", "unknown", "disconnected", "finished", "removed", "deselected"])
|
|
|
+ async def test_poll_cancels_stale_inspection_without_waiting_for_response(
|
|
|
+ self, detector, http_client, manager, clock, change
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings(action="pause"))
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+ processing = asyncio.Event()
|
|
|
+
|
|
|
+ async def pending_response(url, **kwargs):
|
|
|
+ processing.set()
|
|
|
+ await asyncio.Event().wait()
|
|
|
+
|
|
|
+ http_client.post.side_effect = pending_response
|
|
|
+ clock.return_value = 140
|
|
|
+ await detector._poll_once(settings(action="pause"))
|
|
|
+ check = detector._checks[1]
|
|
|
+ try:
|
|
|
+ await asyncio.wait_for(processing.wait(), timeout=1)
|
|
|
+ configuration = settings(action="pause")
|
|
|
+ if change in ("paused", "unknown", "finished"):
|
|
|
+ manager.get_all_statuses.return_value[1] = printer_status(
|
|
|
+ state={"paused": "PAUSE", "unknown": "unknown", "finished": "FINISH"}[change]
|
|
|
+ )
|
|
|
+ elif change == "disconnected":
|
|
|
+ manager.is_connected.return_value = False
|
|
|
+ elif change == "removed":
|
|
|
+ manager.get_all_statuses.return_value = {}
|
|
|
+ else:
|
|
|
+ configuration["enabled_printers"] = set()
|
|
|
+ await detector._poll_once(configuration)
|
|
|
+ await asyncio.wait_for(asyncio.gather(check, return_exceptions=True), timeout=1)
|
|
|
+
|
|
|
+ assert check.cancelled()
|
|
|
+ assert detector._checks == {}
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+ if change in ("paused", "unknown", "disconnected"):
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.frame_count == 1
|
|
|
+ else:
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+ finally:
|
|
|
+ detector.stop()
|
|
|
+ await asyncio.gather(check, return_exceptions=True)
|
|
|
+
|
|
|
+ async def test_completed_old_check_cannot_remove_replacement(self, detector, manager, clock):
|
|
|
+ started = asyncio.Event()
|
|
|
+
|
|
|
+ async def capture(printer_id):
|
|
|
+ started.set()
|
|
|
+ await asyncio.Event().wait()
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ await detector._poll_once(settings())
|
|
|
+ old_check = detector._checks[1]
|
|
|
+ await asyncio.wait_for(started.wait(), timeout=1)
|
|
|
+ manager.get_all_statuses.return_value[1] = printer_status(subtask_id="print-2")
|
|
|
+ await detector._poll_once(settings())
|
|
|
+ new_check = detector._checks[1]
|
|
|
+ try:
|
|
|
+ await asyncio.wait_for(asyncio.gather(old_check, return_exceptions=True), timeout=1)
|
|
|
+ assert old_check.cancelled()
|
|
|
+ assert new_check is not old_check
|
|
|
+ assert detector._checks[1] is new_check
|
|
|
+ assert not new_check.done()
|
|
|
+ assert detector._states[1].print_id == "print-2"
|
|
|
+ finally:
|
|
|
+ detector.stop()
|
|
|
+ await asyncio.gather(old_check, new_check, return_exceptions=True)
|
|
|
+
|
|
|
+ async def test_unexpected_check_error_is_logged_safely_and_does_not_block_retry(self, detector, manager, caplog):
|
|
|
+ detector._check_printer = AsyncMock(side_effect=[RuntimeError("secret-api-key"), None])
|
|
|
+ with caplog.at_level(logging.ERROR, logger=MODULE):
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._checks == {}
|
|
|
+ await poll_once(detector, settings())
|
|
|
+
|
|
|
+ assert detector._check_printer.await_count == 2
|
|
|
+ assert "OctoEverywhere printer check failed: RuntimeError" in caplog.text
|
|
|
+ assert "secret-api-key" not in caplog.text
|
|
|
+
|
|
|
+
|
|
|
+class TestLifecycle:
|
|
|
+ @pytest.mark.parametrize("initial_id", [None, "", "0", 0])
|
|
|
+ async def test_late_job_id_reuses_context_and_later_new_job_replaces_it(
|
|
|
+ self, detector, http_client, manager, clock, initial_id
|
|
|
+ ):
|
|
|
+ new_url = PRIMARY_URL.replace("context-one", "context-two")
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="context-two", ProcessRequestUrl=new_url)),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ status = manager.get_all_statuses.return_value[1]
|
|
|
+ status.subtask_id = initial_id
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+
|
|
|
+ status.subtask_id = "print-1"
|
|
|
+ clock.return_value = 140
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.frame_count == 2
|
|
|
+
|
|
|
+ status.subtask_id = "print-2"
|
|
|
+ clock.return_value = 180
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is not state
|
|
|
+ assert detector._states[1].context.context_id == "context-two"
|
|
|
+ assert detector._states[1].frame_count == 1
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ new_url,
|
|
|
+ ]
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("missing_id", [None, "", "0", 0])
|
|
|
+ async def test_missing_job_id_preserves_context_and_last_known_job(
|
|
|
+ self, detector, http_client, manager, clock, missing_id
|
|
|
+ ):
|
|
|
+ new_url = PRIMARY_URL.replace("context-one", "context-two")
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="context-two", ProcessRequestUrl=new_url)),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ status = manager.get_all_statuses.return_value[1]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+
|
|
|
+ for now, job_id in ((140, missing_id), (180, "print-1"), (220, missing_id)):
|
|
|
+ status.subtask_id = job_id
|
|
|
+ clock.return_value = now
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.frame_count == 4
|
|
|
+
|
|
|
+ status.subtask_id = "print-2"
|
|
|
+ clock.return_value = 260
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is not state
|
|
|
+ assert detector._states[1].context.context_id == "context-two"
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ new_url,
|
|
|
+ ]
|
|
|
+
|
|
|
+ async def test_filename_and_name_changes_without_job_id_reuse_printer_context(
|
|
|
+ self, detector, http_client, manager, clock
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ status = manager.get_all_statuses.return_value[1]
|
|
|
+ status.subtask_id = None
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+
|
|
|
+ for now, name, filename in ((140, "renamed-print", "plate_2.gcode"), (180, None, None)):
|
|
|
+ status.subtask_name = name
|
|
|
+ status.gcode_file = filename
|
|
|
+ clock.return_value = now
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.frame_count == 3
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ ]
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "transient_status",
|
|
|
+ [printer_status(state="unknown"), printer_status(state="PREPARE"), printer_status(state=None), None],
|
|
|
+ )
|
|
|
+ async def test_transient_printer_status_preserves_context_without_uploading(
|
|
|
+ self, detector, http_client, manager, clock, transient_status
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ status = manager.get_all_statuses.return_value[1]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+
|
|
|
+ manager.get_all_statuses.return_value[1] = transient_status
|
|
|
+ clock.return_value = 300
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ detector._capture_frame.assert_awaited_once()
|
|
|
+
|
|
|
+ manager.get_all_statuses.return_value[1] = status
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.frame_count == 2
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ ]
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("phase", ["capture", "context", "process"])
|
|
|
+ async def test_late_metadata_during_inspection_preserves_context_and_result(
|
|
|
+ self, detector, http_client, clock, phase
|
|
|
+ ):
|
|
|
+ status = printer_status(subtask_id=None)
|
|
|
+
|
|
|
+ def update_metadata():
|
|
|
+ status.subtask_id = "print-1"
|
|
|
+ status.subtask_name = "updated-print-name"
|
|
|
+ status.gcode_file = "updated-plate.gcode"
|
|
|
+
|
|
|
+ async def capture(printer_id):
|
|
|
+ if phase == "capture":
|
|
|
+ update_metadata()
|
|
|
+ return FAKE_JPEG
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ if phase == "context":
|
|
|
+ update_metadata()
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ if phase == "process":
|
|
|
+ update_metadata()
|
|
|
+ return httpx.Response(200, json=process_payload())
|
|
|
+
|
|
|
+ detector._capture_frame.side_effect = capture
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, status, settings())
|
|
|
+ state = detector._states[1]
|
|
|
+ context = state.context
|
|
|
+ assert context is not None
|
|
|
+ assert state.frame_count == 1
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+
|
|
|
+ clock.return_value = 140
|
|
|
+ await detector._check_printer(1, status, settings())
|
|
|
+ assert detector._states[1] is state
|
|
|
+ assert state.context is context
|
|
|
+ assert state.frame_count == 2
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ ]
|
|
|
+
|
|
|
+ async def test_known_job_change_during_processing_discards_stale_result(self, detector, http_client, clock):
|
|
|
+ status = printer_status()
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ status.subtask_id = "print-2"
|
|
|
+ return httpx.Response(200, json=process_payload(PauseSuggested=True))
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await detector._check_printer(1, status, settings(action="pause"))
|
|
|
+ state = detector._states[1]
|
|
|
+ assert state.frame_count == 0
|
|
|
+ assert detector.get_status()["history"] == []
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ new_url = PRIMARY_URL.replace("context-one", "context-two")
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="context-two", ProcessRequestUrl=new_url)),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await detector._check_printer(1, status, settings(action="pause"))
|
|
|
+ assert detector._states[1] is not state
|
|
|
+ assert detector._states[1].context.context_id == "context-two"
|
|
|
+ assert detector._states[1].frame_count == 1
|
|
|
+ assert [call.args[0] for call in http_client.post.await_args_list] == [
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ PRIMARY_URL,
|
|
|
+ CREATE_CONTEXT_URL,
|
|
|
+ new_url,
|
|
|
+ ]
|
|
|
+ detector._dispatch_action.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_pause_preserves_context_and_does_not_upload(self, detector, http_client, manager, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ context = detector._states[1].context
|
|
|
+ manager.get_all_statuses.return_value[1].state = "PAUSE"
|
|
|
+ clock.return_value = 300
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ manager.get_all_statuses.return_value[1].state = "RUNNING"
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("terminal_state", ["FINISH", "FAILED", "IDLE"])
|
|
|
+ async def test_end_clears_state_and_identical_job_gets_new_context(
|
|
|
+ self, detector, http_client, manager, clock, terminal_state
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ manager.get_all_statuses.return_value[1].state = terminal_state
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+ manager.get_all_statuses.return_value[1].state = "RUNNING"
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1].context.context_id == "new-context"
|
|
|
+ assert detector._states[1].frame_count == 1
|
|
|
+
|
|
|
+ async def test_same_filename_new_job_id_gets_fresh_context(self, detector, http_client, manager, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True)),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True)),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ manager.get_all_statuses.return_value[1].subtask_id = "print-2"
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1].context.context_id == "new-context"
|
|
|
+ assert detector._dispatch_action.await_count == 2
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("changes", [{"api_key": "replacement-key"}, {"confidence": "high"}])
|
|
|
+ async def test_credentials_or_confidence_change_recreates_context(
|
|
|
+ self, detector, http_client, manager, clock, changes
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True)),
|
|
|
+ httpx.Response(200, json=context_payload(ContextId="new-context")),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True)),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ await poll_once(detector, settings(**changes))
|
|
|
+ assert detector._states[1].context is None
|
|
|
+ assert detector._states[1].next_check_at == 140
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ clock.return_value = 140
|
|
|
+ await poll_once(detector, settings(**changes))
|
|
|
+ assert detector._states[1].context.context_id == "new-context"
|
|
|
+ detector._dispatch_action.assert_awaited_once()
|
|
|
+ creation = http_client.post.await_args_list[2]
|
|
|
+ assert creation.kwargs["headers"]["X-API-Key"] == changes.get("api_key", "secret-api-key")
|
|
|
+ assert creation.kwargs["json"]["PauseConfidenceLevel"] == (4 if "confidence" in changes else 3)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "changes", [{"enabled": False}, {"api_key": ""}, {"enabled_printers": set()}, {"enabled_printers": {2}}]
|
|
|
+ )
|
|
|
+ async def test_disabled_printers_stop_uploading_and_clear_status(
|
|
|
+ self, detector, http_client, manager, clock, changes
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ await poll_once(detector, settings(**changes))
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+
|
|
|
+ async def test_deleted_printer_clears_state(self, detector, http_client, manager, clock):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ manager.get_all_statuses.return_value = {}
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+
|
|
|
+ async def test_disconnected_printer_retains_context_without_stale_safe_status(
|
|
|
+ self, detector, http_client, manager, clock
|
|
|
+ ):
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ httpx.Response(200, json=process_payload()),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ context = detector._states[1].context
|
|
|
+ manager.is_connected.return_value = False
|
|
|
+ clock.return_value = 300
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "error"
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+ manager.is_connected.return_value = True
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert detector._states[1].context is context
|
|
|
+ assert detector.get_per_printer()[1]["class"] == "safe"
|
|
|
+
|
|
|
+ async def test_each_printer_owns_a_separate_context_and_interval(self, detector, http_client, manager, clock):
|
|
|
+ manager.get_all_statuses.return_value[2] = printer_status(subtask_id="print-2")
|
|
|
+ count = 0
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ nonlocal count
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ count += 1
|
|
|
+ return httpx.Response(200, json=context_payload(ContextId=f"context-{count}"))
|
|
|
+ return httpx.Response(
|
|
|
+ 200, json=process_payload(NextProcessIntervalSec={"Minimum": 40 * count, "Recommended": 20})
|
|
|
+ )
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ await poll_once(detector, settings())
|
|
|
+ assert {state.context.context_id for state in detector._states.values()} == {"context-1", "context-2"}
|
|
|
+ assert detector._states[1].next_check_at == 140
|
|
|
+ assert detector._states[2].next_check_at == 180
|
|
|
+
|
|
|
+ def test_stop_cancels_service_and_clears_live_status(self, detector):
|
|
|
+ task = MagicMock()
|
|
|
+ detector._task = task
|
|
|
+ detector._last_error = "an old error"
|
|
|
+ detector.stop()
|
|
|
+ task.cancel.assert_called_once()
|
|
|
+ assert detector.get_status() == {
|
|
|
+ "is_running": False,
|
|
|
+ "last_error": None,
|
|
|
+ "last_error_code": None,
|
|
|
+ "per_printer": {},
|
|
|
+ "history": [],
|
|
|
+ }
|
|
|
+
|
|
|
+ async def test_enabled_without_key_reports_configuration_error(self, detector, http_client, manager):
|
|
|
+ await poll_once(detector, settings(api_key=""))
|
|
|
+ assert "API key is required" in detector.get_status()["last_error"]
|
|
|
+ http_client.post.assert_not_awaited()
|
|
|
+
|
|
|
+ async def test_start_is_idempotent(self, detector):
|
|
|
+ with patch.object(detector, "_loop", new=AsyncMock()):
|
|
|
+ await detector.start()
|
|
|
+ first_task = detector._task
|
|
|
+ await detector.start()
|
|
|
+ assert detector._task is first_task
|
|
|
+ assert detector.get_status()["is_running"] is True
|
|
|
+ detector.stop()
|
|
|
+ with pytest.raises(asyncio.CancelledError):
|
|
|
+ await first_task
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "enabled, poll_interval, expected_poll", [(False, 20, 5), (True, 5, 1), (True, 21, 1), (True, 30, 1)]
|
|
|
+ )
|
|
|
+ async def test_enabled_detection_uses_one_second_scheduler_resolution(
|
|
|
+ self, detector, enabled, poll_interval, expected_poll
|
|
|
+ ):
|
|
|
+ detector._load_settings = AsyncMock(return_value=settings(enabled=enabled, poll_interval=poll_interval))
|
|
|
+ detector._poll_once = AsyncMock()
|
|
|
+ with patch(f"{MODULE}.asyncio.sleep", new=AsyncMock(side_effect=asyncio.CancelledError)) as sleep:
|
|
|
+ await detector._loop()
|
|
|
+ sleep.assert_awaited_once_with(expected_poll)
|
|
|
+
|
|
|
+ async def test_refresh_cancels_stale_settings_load_before_polling(self, detector, clock):
|
|
|
+ loading = asyncio.Event()
|
|
|
+ cancelled = asyncio.Event()
|
|
|
+
|
|
|
+ async def load_old_settings():
|
|
|
+ loading.set()
|
|
|
+ try:
|
|
|
+ await asyncio.Event().wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ cancelled.set()
|
|
|
+ raise
|
|
|
+
|
|
|
+ detector._load_settings = AsyncMock(side_effect=load_old_settings)
|
|
|
+ detector._poll_once = AsyncMock()
|
|
|
+ await detector.start()
|
|
|
+ await asyncio.wait_for(loading.wait(), timeout=1)
|
|
|
+ detector._load_settings = AsyncMock(return_value=settings(enabled=False))
|
|
|
+ await detector.refresh_settings()
|
|
|
+ assert cancelled.is_set()
|
|
|
+ assert detector.get_status()["is_running"] is True
|
|
|
+ await asyncio.sleep(0)
|
|
|
+ for call in detector._poll_once.await_args_list:
|
|
|
+ assert call.args[0]["enabled"] is False
|
|
|
+ task = detector._task
|
|
|
+ detector.stop()
|
|
|
+ await asyncio.gather(task, return_exceptions=True)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("change", ["print_ended", "disabled", "key_changed", "account_blocked"])
|
|
|
+ async def test_cancel_during_action_lookup_prevents_hardware_control(
|
|
|
+ self, detector, http_client, manager, clock, change
|
|
|
+ ):
|
|
|
+ lookup_started = asyncio.Event()
|
|
|
+ lookup_cancelled = asyncio.Event()
|
|
|
+
|
|
|
+ async def get_printer_name(printer_id):
|
|
|
+ lookup_started.set()
|
|
|
+ try:
|
|
|
+ await asyncio.Event().wait()
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ lookup_cancelled.set()
|
|
|
+ raise
|
|
|
+
|
|
|
+ # Exercise the real action dispatcher: its printer-name database lookup
|
|
|
+ # awaits before pause/power-off, which used to leave a cancellation gap.
|
|
|
+ detector._dispatch_action = OctoEverywhereDetectionService._dispatch_action.__get__(detector)
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(PauseSuggested=True)),
|
|
|
+ ]
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.obico_actions._get_printer_name", side_effect=get_printer_name),
|
|
|
+ patch("backend.app.services.obico_actions._pause_print") as pause,
|
|
|
+ patch("backend.app.services.obico_actions._turn_off_linked_plugs", new=AsyncMock()) as turn_off,
|
|
|
+ patch("backend.app.services.obico_actions._notify", new=AsyncMock()) as notify,
|
|
|
+ ):
|
|
|
+ poll = asyncio.create_task(poll_once(detector, settings(action="pause_and_off")))
|
|
|
+ await asyncio.wait_for(lookup_started.wait(), timeout=1)
|
|
|
+ if change == "print_ended":
|
|
|
+ detector.reset_printer(1)
|
|
|
+ elif change == "account_blocked":
|
|
|
+ http_client.post.side_effect = [httpx.Response(200, json=context_payload()), quota_response()]
|
|
|
+ await detector._check_printer(2, printer_status(subtask_id="print-2"), settings(action="pause_and_off"))
|
|
|
+ else:
|
|
|
+ detector._load_settings = AsyncMock(
|
|
|
+ return_value=settings(enabled=False)
|
|
|
+ if change == "disabled"
|
|
|
+ else settings(api_key="replacement-key")
|
|
|
+ )
|
|
|
+ await detector.refresh_settings()
|
|
|
+ await asyncio.wait_for(poll, timeout=1)
|
|
|
+ assert lookup_cancelled.is_set()
|
|
|
+ pause.assert_not_called()
|
|
|
+ turn_off.assert_not_awaited()
|
|
|
+ notify.assert_not_awaited()
|
|
|
+ if change == "key_changed":
|
|
|
+ assert detector._states[1].action_fired is False
|
|
|
+ assert detector._states[1].warning_fired is False
|
|
|
+ assert detector._states[1].context is None
|
|
|
+ elif change == "account_blocked":
|
|
|
+ assert detector.get_status()["last_error_code"] == USAGE_LIMIT_ERROR
|
|
|
+ assert detector._states[1].action_fired is False
|
|
|
+ assert detector._states[1].warning_fired is False
|
|
|
+ else:
|
|
|
+ assert detector.get_per_printer() == {}
|
|
|
+
|
|
|
+ async def test_cancelled_request_preserves_deadline_on_key_change(self, detector, http_client, manager, clock):
|
|
|
+ processing = asyncio.Event()
|
|
|
+
|
|
|
+ async def respond(url, **kwargs):
|
|
|
+ if url == CREATE_CONTEXT_URL:
|
|
|
+ return httpx.Response(200, json=context_payload())
|
|
|
+ processing.set()
|
|
|
+ await asyncio.Event().wait()
|
|
|
+
|
|
|
+ http_client.post.side_effect = respond
|
|
|
+ poll = asyncio.create_task(poll_once(detector, settings()))
|
|
|
+ await asyncio.wait_for(processing.wait(), timeout=1)
|
|
|
+ clock.return_value = 120
|
|
|
+ detector._load_settings = AsyncMock(return_value=settings(api_key="replacement-key"))
|
|
|
+ await detector.refresh_settings()
|
|
|
+ await asyncio.wait_for(poll, timeout=1)
|
|
|
+ assert detector._states[1].next_check_at == 140
|
|
|
+ assert detector._states[1].context is None
|
|
|
+ await poll_once(detector, settings(api_key="replacement-key"))
|
|
|
+ assert http_client.post.await_count == 2
|
|
|
+
|
|
|
+ async def test_interval_edit_during_request_preserves_server_minimum(self, detector, http_client, manager, clock):
|
|
|
+ processing = asyncio.Event()
|
|
|
+
|
|
|
+ async def pending_response(url, **kwargs):
|
|
|
+ processing.set()
|
|
|
+ await asyncio.Event().wait()
|
|
|
+
|
|
|
+ http_client.post.side_effect = [
|
|
|
+ httpx.Response(200, json=context_payload()),
|
|
|
+ httpx.Response(200, json=process_payload(NextProcessIntervalSec={"Minimum": 60, "Recommended": 20})),
|
|
|
+ ]
|
|
|
+ await poll_once(detector, settings(poll_interval=30))
|
|
|
+ http_client.post.side_effect = pending_response
|
|
|
+ clock.return_value = 160
|
|
|
+ poll = asyncio.create_task(poll_once(detector, settings(poll_interval=30)))
|
|
|
+ await asyncio.wait_for(processing.wait(), timeout=1)
|
|
|
+ clock.return_value = 165
|
|
|
+ detector._load_settings = AsyncMock(return_value=settings(poll_interval=5))
|
|
|
+ await detector.refresh_settings()
|
|
|
+ await asyncio.wait_for(poll, timeout=1)
|
|
|
+ assert detector._states[1].next_check_at == 225
|
|
|
+ clock.return_value = 224
|
|
|
+ await poll_once(detector, settings(poll_interval=5))
|
|
|
+ assert http_client.post.await_count == 3
|
|
|
+
|
|
|
+
|
|
|
+class TestSettings:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "raw, expected", [("", None), ("[1, 3]", {1, 3}), ("[]", set()), ("bad", set()), ("{}", set())]
|
|
|
+ )
|
|
|
+ async def test_loads_settings_and_printer_selection(self, detector, raw, expected):
|
|
|
+ rows = [
|
|
|
+ SimpleNamespace(key="octoeverywhere_enabled", value="true"),
|
|
|
+ SimpleNamespace(key="octoeverywhere_api_key", value=" secret-api-key "),
|
|
|
+ SimpleNamespace(key="octoeverywhere_enabled_printers", value=raw),
|
|
|
+ ]
|
|
|
+ db = AsyncMock()
|
|
|
+ result = MagicMock()
|
|
|
+ result.scalars.return_value.all.return_value = rows
|
|
|
+ db.execute.return_value = result
|
|
|
+ with patch(f"{MODULE}.async_session") as session:
|
|
|
+ session.return_value.__aenter__.return_value = db
|
|
|
+ loaded = await detector._load_settings()
|
|
|
+
|
|
|
+ assert loaded == settings(enabled_printers=expected)
|
|
|
+
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "raw, expected",
|
|
|
+ [(None, 20), ("", 20), ("null", 20), ("5", 5), ("20", 20), ("21", 21), ("30", 30), ("4", 20), ("31", 20)],
|
|
|
+ )
|
|
|
+ async def test_loads_inspection_interval(self, detector, raw, expected):
|
|
|
+ db = AsyncMock()
|
|
|
+ result = MagicMock()
|
|
|
+ result.scalars.return_value.all.return_value = [SimpleNamespace(key="octoeverywhere_poll_interval", value=raw)]
|
|
|
+ db.execute.return_value = result
|
|
|
+ with patch(f"{MODULE}.async_session") as session:
|
|
|
+ session.return_value.__aenter__.return_value = db
|
|
|
+ loaded = await detector._load_settings()
|
|
|
+ assert loaded["poll_interval"] == expected
|