From af5b9fe00720eed98a14a57e820733bd8bab68ef Mon Sep 17 00:00:00 2001 From: Nic Boet Date: Thu, 12 Feb 2026 00:11:05 -0600 Subject: [PATCH 1/4] Add test suite with 72 tests covering core API surface Introduce pytest-based test infrastructure for pyzm, which previously had zero tests. Establishes a safety net ahead of upcoming refactoring efforts. - pytest config (pyproject.toml) with markers and coverage settings - JSON response fixtures mirroring real ZM API payloads - Shared conftest fixtures (mock login, suppressed logger, exit guard) - Unit tests: auth flows, API methods, request handling, helper edge cases - Integration tests: full login -> monitors -> events -> states workflows - Testing strategy documentation (docs/testing-strategy.md) Core modules coverage: api.py 91%, helpers 80-100%. Co-Authored-By: Claude Opus 4.6 --- .gitignore | 3 + pyproject.toml | 26 ++ tests/README.md | 126 +++++++ tests/__init__.py | 0 tests/conftest.py | 178 +++++++++ tests/fixtures/responses/configs.json | 36 ++ tests/fixtures/responses/daemon_status.json | 4 + tests/fixtures/responses/events.json | 54 +++ tests/fixtures/responses/login_legacy.json | 6 + tests/fixtures/responses/login_success.json | 10 + tests/fixtures/responses/monitors.json | 34 ++ tests/fixtures/responses/states.json | 28 ++ tests/fixtures/responses/version.json | 4 + tests/integration/__init__.py | 0 tests/integration/test_api_workflow.py | 166 +++++++++ tests/requirements-test.txt | 4 + tests/unit/__init__.py | 0 tests/unit/helpers/__init__.py | 0 tests/unit/helpers/test_base.py | 60 +++ tests/unit/helpers/test_events.py | 131 +++++++ tests/unit/helpers/test_monitor.py | 118 ++++++ tests/unit/helpers/test_state.py | 35 ++ tests/unit/helpers/test_states.py | 68 ++++ tests/unit/test_api_auth.py | 294 +++++++++++++++ tests/unit/test_api_methods.py | 388 ++++++++++++++++++++ tests/unit/test_api_request.py | 289 +++++++++++++++ 26 files changed, 2062 insertions(+) create mode 100644 pyproject.toml create mode 100644 tests/README.md create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/fixtures/responses/configs.json create mode 100644 tests/fixtures/responses/daemon_status.json create mode 100644 tests/fixtures/responses/events.json create mode 100644 tests/fixtures/responses/login_legacy.json create mode 100644 tests/fixtures/responses/login_success.json create mode 100644 tests/fixtures/responses/monitors.json create mode 100644 tests/fixtures/responses/states.json create mode 100644 tests/fixtures/responses/version.json create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/test_api_workflow.py create mode 100644 tests/requirements-test.txt create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/helpers/__init__.py create mode 100644 tests/unit/helpers/test_base.py create mode 100644 tests/unit/helpers/test_events.py create mode 100644 tests/unit/helpers/test_monitor.py create mode 100644 tests/unit/helpers/test_state.py create mode 100644 tests/unit/helpers/test_states.py create mode 100644 tests/unit/test_api_auth.py create mode 100644 tests/unit/test_api_methods.py create mode 100644 tests/unit/test_api_request.py diff --git a/.gitignore b/.gitignore index fe4d0fe..46d70fd 100644 --- a/.gitignore +++ b/.gitignore @@ -14,4 +14,7 @@ __pycache__/ # due to using tox and pytest .tox .cache +.pytest_cache/ +.coverage +htmlcov/ .idea/* \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a782fca --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[tool.pytest.ini_options] +testpaths = ["tests"] +markers = [ + "unit: Unit tests (fast, no external deps)", + "integration: Integration tests (multi-step workflows)", + "slow: Slow tests", +] +addopts = "-v --strict-markers --tb=short" + +[tool.coverage.run] +source = ["pyzm"] +omit = [ + "pyzm/ml/*", + "pyzm/ZMLog.py", + "pyzm/ZMMemory.py", + "pyzm/ZMEventNotification.py", + "pyzm/helpers/Media.py", + "pyzm/helpers/utils.py", +] + +[tool.coverage.report] +show_missing = true +exclude_lines = [ + "pragma: no cover", + "if __name__ == .__main__.", +] diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..d201a3c --- /dev/null +++ b/tests/README.md @@ -0,0 +1,126 @@ +# pyzm Testing Strategy + +## Philosophy + +pyzm testing follows a **public API first** approach: + +- **Public API tests** (`tests/unit/test_api_*.py`) test `ZMApi` methods as consumers use them. All HTTP is mocked via `responses`. These are the **primary coverage layer** — since `zm_api.monitors()` creates `Monitors` → `Monitor` objects from API data, the happy path for helper accessors is already covered here. +- **Helper edge-case tests** (`tests/unit/helpers/`) cover ONLY edge cases, pure logic, and error paths not reachable through the public API. Examples: `States.find()` case-insensitive search, `Monitor.set_parameter()` payload construction. +- **Integration tests** (`tests/integration/`) chain multiple API calls in realistic workflows. + +**Rule: if an API test already asserts a helper behavior, don't write a separate helper test for it.** + +## Directory Structure + +``` +tests/ +├── conftest.py # Shared fixtures +├── fixtures/ +│ └── responses/ # JSON response fixtures mirroring ZM API payloads +│ ├── login_success.json +│ ├── login_legacy.json +│ ├── monitors.json +│ ├── events.json +│ ├── states.json +│ ├── configs.json +│ ├── version.json +│ └── daemon_status.json +├── unit/ +│ ├── test_api_auth.py # Login flows, token refresh, relogin +│ ├── test_api_methods.py # monitors(), events(), states(), configs(), etc. +│ ├── test_api_request.py # _make_request retry, error handling, content types +│ └── helpers/ +│ ├── test_monitor.py # set_parameter payload, arm/disarm URLs +│ ├── test_events.py # URL filter building, pagination +│ ├── test_state.py # active() false, definition() None +│ ├── test_states.py # find() search logic +│ └── test_base.py # ConsoleLog level filtering, exit calls +└── integration/ + └── test_api_workflow.py # Full login -> monitors -> events -> states +``` + +## Fixture Patterns + +### JSON Response Fixtures + +Response fixtures in `tests/fixtures/responses/` mirror actual ZM API payloads. They are loaded by helper functions in `conftest.py` and provided as pytest fixtures. + +### Key Shared Fixtures + +| Fixture | Purpose | +|---|---| +| `zm_options` | Standard config dict for JWT auth | +| `zm_options_no_auth` | Config without credentials | +| `zm_api` | Pre-authenticated `ZMApi` with JWT login mocked | +| `zm_api_legacy` | Pre-authenticated `ZMApi` with legacy credentials | +| `suppress_logger` | (autouse) Patches `g.logger` to silent mock | +| `no_exit` | (autouse) Patches `builtins.exit` | + +### Test Isolation + +Each test that makes HTTP calls uses either: +- `@responses.activate` decorator for full control +- The `zm_api` fixture which uses `responses.RequestsMock` context manager + +The `suppress_logger` and `no_exit` fixtures are `autouse=True` — they apply to all tests automatically. + +## Mocking Strategy + +### HTTP Mocking with `responses` + +We use the `responses` library (not `requests-mock` or VCR): +- Simple decorator/context manager API +- No cassettes to maintain +- Best ergonomics for `requests`-based code + +### Why Not VCR? + +VCR records real HTTP interactions. We don't have a live ZM server in CI, and maintaining cassettes adds complexity without value here. + +### Global State + +pyzm uses a global logger at `pyzm.helpers.globals.logger`. The `suppress_logger` autouse fixture replaces it with a `MagicMock` for every test, preventing console spam and avoiding test interdependence. + +## Running Tests + +```bash +# All tests +pytest tests/ -v + +# With coverage +pytest tests/ -v --cov=pyzm --cov-report=term-missing + +# Only unit tests +pytest tests/unit/ -v + +# Only integration tests +pytest tests/integration/ -v -m integration + +# Specific test file +pytest tests/unit/test_api_auth.py -v +``` + +## Coverage Targets + +- Core modules (`api.py`, helpers): 80%+ +- Excluded from coverage: `pyzm/ml/*`, `ZMLog.py`, `ZMMemory.py`, `ZMEventNotification.py`, `helpers/Media.py`, `helpers/utils.py` + +## Testing Challenges Specific to pyzm + +| Challenge | Solution | +|---|---| +| `ZMApi.__init__` calls `_login()` | Every test creating a `ZMApi` must mock the login endpoint via `responses` | +| `g.logger` global mutable state | `conftest.py` autouse fixture patches it to a silent mock | +| `ConsoleLog.Fatal()`/`Panic()` call `exit()` | Autouse fixture patches `builtins.exit` | +| `options={}` mutable default args | Tests always pass `.copy()` of options dicts | +| `Event.py` imports `progressbar` | Module-level import; `progressbar2` in test deps | +| `utils.py` imports `cv2`/`numpy` | Excluded from coverage; tested separately if needed | + +## What We Explicitly Do NOT Test (Yet) + +- `pyzm.ml.*` — ML modules with heavy deps (cv2, TensorFlow, dlib) +- `pyzm.ZMLog` — Requires MySQL connection +- `pyzm.ZMMemory` — Requires shared memory segments +- `pyzm.ZMEventNotification` — Requires WebSocket +- `pyzm.helpers.Media` — Requires cv2/numpy +- `pyzm.helpers.utils` — `draw_bbox` requires cv2; `Timer`/`read_config`/`template_fill` are simple utilities diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..0ea1c14 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,178 @@ +"""Shared test fixtures for pyzm test suite.""" + +import json +import os +from unittest.mock import MagicMock, patch + +import pytest +import responses + +FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures", "responses") + + +def _load_fixture(name): + with open(os.path.join(FIXTURES_DIR, name)) as f: + return json.load(f) + + +# --------------------------------------------------------------------------- +# Option dicts +# --------------------------------------------------------------------------- + +@pytest.fixture +def zm_options(): + """Standard ZM API options dict for JWT auth.""" + return { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "user": "admin", + "password": "secret", + "disable_ssl_cert_check": True, + } + + +@pytest.fixture +def zm_options_no_auth(): + """ZM API options without authentication.""" + return { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "disable_ssl_cert_check": True, + } + + +@pytest.fixture +def zm_options_token(): + """ZM API options with token auth.""" + return { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "user": "admin", + "password": "secret", + "token": "existing_token_123", + "disable_ssl_cert_check": True, + } + + +# --------------------------------------------------------------------------- +# JSON response fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def login_success_response(): + return _load_fixture("login_success.json") + + +@pytest.fixture +def login_legacy_response(): + return _load_fixture("login_legacy.json") + + +@pytest.fixture +def monitors_response(): + return _load_fixture("monitors.json") + + +@pytest.fixture +def events_response(): + return _load_fixture("events.json") + + +@pytest.fixture +def states_response(): + return _load_fixture("states.json") + + +@pytest.fixture +def configs_response(): + return _load_fixture("configs.json") + + +@pytest.fixture +def version_response(): + return _load_fixture("version.json") + + +@pytest.fixture +def daemon_status_response(): + return _load_fixture("daemon_status.json") + + +# --------------------------------------------------------------------------- +# Logger suppression +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def suppress_logger(): + """Replace g.logger with a silent mock to prevent console spam.""" + mock_logger = MagicMock() + with patch("pyzm.helpers.globals.logger", mock_logger): + # Also patch the module-level reference that may have been cached + import pyzm.helpers.globals as g + original = g.logger + g.logger = mock_logger + yield mock_logger + g.logger = original + + +# --------------------------------------------------------------------------- +# Prevent exit() calls in ConsoleLog.Fatal / Panic +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def no_exit(): + """Prevent ConsoleLog.Fatal/Panic from killing the test runner.""" + with patch("builtins.exit") as mock_exit: + yield mock_exit + + +# --------------------------------------------------------------------------- +# Pre-authenticated ZMApi factory +# --------------------------------------------------------------------------- + +@pytest.fixture +def zm_api(zm_options, login_success_response): + """Return a pre-authenticated ZMApi instance with login mocked.""" + from pyzm.api import ZMApi + + with responses.RequestsMock() as rsps: + rsps.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + api = ZMApi(options=zm_options.copy()) + + assert api.authenticated is True + assert api.api_version == "2.0" + assert api.access_token == "test_access_token_abc123" + return api + + +@pytest.fixture +def zm_api_legacy(login_legacy_response): + """Return a pre-authenticated ZMApi instance using legacy credentials.""" + from pyzm.api import ZMApi + + options = { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "user": "admin", + "password": "secret", + "disable_ssl_cert_check": True, + } + + with responses.RequestsMock() as rsps: + rsps.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_legacy_response, + status=200, + ) + api = ZMApi(options=options.copy()) + + assert api.authenticated is True + assert api.api_version == "1.0" + assert api.legacy_credentials == "auth=abc123hash" + return api diff --git a/tests/fixtures/responses/configs.json b/tests/fixtures/responses/configs.json new file mode 100644 index 0000000..59fea53 --- /dev/null +++ b/tests/fixtures/responses/configs.json @@ -0,0 +1,36 @@ +{ + "configs": [ + { + "Config": { + "Id": "1", + "Name": "ZM_AUTH_TYPE", + "Value": "builtin", + "Type": "string", + "DefaultValue": "builtin", + "Prompt": "Authentication type", + "Help": "Specifies which authentication method is used", + "Category": "System", + "Readonly": "0", + "Requires": null + } + }, + { + "Config": { + "Id": "2", + "Name": "ZM_AUTH_HASH_LOGINS", + "Value": "1", + "Type": "boolean", + "DefaultValue": "0" + } + }, + { + "Config": { + "Id": "3", + "Name": "ZM_AUTH_HASH_SECRET", + "Value": "mysecret", + "Type": "string", + "DefaultValue": "" + } + } + ] +} diff --git a/tests/fixtures/responses/daemon_status.json b/tests/fixtures/responses/daemon_status.json new file mode 100644 index 0000000..f6a0d40 --- /dev/null +++ b/tests/fixtures/responses/daemon_status.json @@ -0,0 +1,4 @@ +{ + "status": true, + "statustext": "Running" +} diff --git a/tests/fixtures/responses/events.json b/tests/fixtures/responses/events.json new file mode 100644 index 0000000..b2bd70e --- /dev/null +++ b/tests/fixtures/responses/events.json @@ -0,0 +1,54 @@ +{ + "events": [ + { + "Event": { + "Id": "100", + "Name": "Event 100", + "MonitorId": "1", + "Cause": "Motion", + "Notes": "detected:person", + "StartTime": "2024-01-15 10:30:00", + "EndTime": "2024-01-15 10:31:00", + "Length": "60.5", + "Frames": "150", + "AlarmFrames": "45", + "TotScore": "500", + "AvgScore": "11.1", + "MaxScore": "85", + "DefaultVideo": "100-video.mp4", + "FileSystemPath": "/var/lib/zoneminder/events/1/2024-01-15/100" + } + }, + { + "Event": { + "Id": "99", + "Name": "Event 99", + "MonitorId": "1", + "Cause": "Motion", + "Notes": "", + "StartTime": "2024-01-15 10:00:00", + "EndTime": "2024-01-15 10:00:30", + "Length": "30.2", + "Frames": "75", + "AlarmFrames": "20", + "TotScore": "200", + "AvgScore": "10.0", + "MaxScore": "50", + "DefaultVideo": null, + "FileSystemPath": "/var/lib/zoneminder/events/1/2024-01-15/99" + } + } + ], + "pagination": { + "page": 1, + "current": 2, + "count": 2, + "prevPage": false, + "nextPage": false, + "pageCount": 1, + "order": null, + "limit": 100, + "options": {}, + "paramType": "named" + } +} diff --git a/tests/fixtures/responses/login_legacy.json b/tests/fixtures/responses/login_legacy.json new file mode 100644 index 0000000..d3fad71 --- /dev/null +++ b/tests/fixtures/responses/login_legacy.json @@ -0,0 +1,6 @@ +{ + "version": "1.32.3", + "apiversion": "1.0", + "credentials": "auth=abc123hash", + "append_password": "0" +} diff --git a/tests/fixtures/responses/login_success.json b/tests/fixtures/responses/login_success.json new file mode 100644 index 0000000..00b9a88 --- /dev/null +++ b/tests/fixtures/responses/login_success.json @@ -0,0 +1,10 @@ +{ + "access_token": "test_access_token_abc123", + "access_token_expires": 3600, + "refresh_token": "test_refresh_token_xyz789", + "refresh_token_expires": 86400, + "version": "1.36.32", + "apiversion": "2.0", + "credentials": null, + "append_password": null +} diff --git a/tests/fixtures/responses/monitors.json b/tests/fixtures/responses/monitors.json new file mode 100644 index 0000000..1e3d4da --- /dev/null +++ b/tests/fixtures/responses/monitors.json @@ -0,0 +1,34 @@ +{ + "monitors": [ + { + "Monitor": { + "Id": "1", + "Name": "Front Door", + "Function": "Modect", + "Enabled": "1", + "Type": "Ffmpeg", + "Width": "1920", + "Height": "1080", + "Protocol": "rtsp", + "Host": "192.168.1.100", + "Port": "554", + "Path": "/stream1" + } + }, + { + "Monitor": { + "Id": "2", + "Name": "Backyard", + "Function": "Monitor", + "Enabled": "0", + "Type": "Ffmpeg", + "Width": "1280", + "Height": "720", + "Protocol": "rtsp", + "Host": "192.168.1.101", + "Port": "554", + "Path": "/stream1" + } + } + ] +} diff --git a/tests/fixtures/responses/states.json b/tests/fixtures/responses/states.json new file mode 100644 index 0000000..1fbd7fc --- /dev/null +++ b/tests/fixtures/responses/states.json @@ -0,0 +1,28 @@ +{ + "states": [ + { + "State": { + "Id": "1", + "Name": "default", + "Definition": "ZM default state", + "IsActive": "1" + } + }, + { + "State": { + "Id": "2", + "Name": "away", + "Definition": "All monitors active", + "IsActive": "0" + } + }, + { + "State": { + "Id": "3", + "Name": "home", + "Definition": "", + "IsActive": "0" + } + } + ] +} diff --git a/tests/fixtures/responses/version.json b/tests/fixtures/responses/version.json new file mode 100644 index 0000000..7087742 --- /dev/null +++ b/tests/fixtures/responses/version.json @@ -0,0 +1,4 @@ +{ + "version": "1.36.32", + "apiversion": "2.0" +} diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_api_workflow.py b/tests/integration/test_api_workflow.py new file mode 100644 index 0000000..958c98f --- /dev/null +++ b/tests/integration/test_api_workflow.py @@ -0,0 +1,166 @@ +"""Integration test — full login -> monitors -> events -> states workflow.""" + +import pytest +import responses + +from pyzm.api import ZMApi + + +@pytest.mark.integration +class TestFullWorkflow: + + @responses.activate + def test_login_monitors_events_states_workflow( + self, + zm_options, + login_success_response, + monitors_response, + events_response, + states_response, + ): + """Full end-to-end flow: login -> get monitors -> get events -> get states -> find active.""" + # 1. Login + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + api = ZMApi(options=zm_options.copy()) + assert api.authenticated is True + + # 2. Get monitors + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + ) + monitors = api.monitors() + assert len(monitors.list()) == 2 + + front_door = monitors.find(name="Front Door") + assert front_door is not None + assert front_door.id() == 1 + assert front_door.enabled() is True + + backyard = monitors.find(name="Backyard") + assert backyard is not None + assert backyard.enabled() is False + + # 3. Get events for a specific monitor + responses.add( + responses.GET, + "https://zm.example.com/zm/api/events/index/MonitorId =:1.json", + json=events_response, + status=200, + ) + events = api.events(options={"mid": 1}) + assert events.count() == 2 + event_list = events.list() + assert event_list[0].id() == 100 + assert event_list[0].monitor_id() == 1 + assert event_list[0].cause() == "Motion" + + # 4. Get states + responses.add( + responses.GET, + "https://zm.example.com/zm/api/states.json", + json=states_response, + status=200, + ) + states = api.states() + state_list = states.list() + assert len(state_list) == 3 + + # 5. Find active state + active_states = [s for s in state_list if s.active()] + assert len(active_states) == 1 + assert active_states[0].name() == "default" + + # Find by name + away = states.find(name="away") + assert away is not None + assert away.active() is False + + @responses.activate + def test_legacy_auth_workflow( + self, + login_legacy_response, + monitors_response, + ): + """Workflow using legacy credentials auth.""" + options = { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "user": "admin", + "password": "secret", + "disable_ssl_cert_check": True, + } + + # Login with legacy credentials + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_legacy_response, + status=200, + ) + api = ZMApi(options=options) + assert api.authenticated is True + assert api.api_version == "1.0" + + # Get monitors — legacy credentials appended to URL + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + match_querystring=False, + ) + monitors = api.monitors() + assert len(monitors.list()) == 2 + + # Verify legacy credentials were in the request URL + request_url = responses.calls[1].request.url + assert "auth=abc123hash" in request_url + + @responses.activate + def test_monitor_operations( + self, + zm_options, + login_success_response, + monitors_response, + ): + """Monitor CRUD operations in a workflow.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + api = ZMApi(options=zm_options.copy()) + + # Get monitors + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + ) + monitors = api.monitors() + mon = monitors.find(id=1) + + # Check version info + version = api.version() + assert version["status"] == "ok" + assert version["api_version"] == "2.0" + + # Get monitor status + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors/daemonStatus/id:1/daemon:zmc.json", + json={"status": True, "statustext": "Running"}, + status=200, + ) + status = mon.status() + assert status["status"] is True diff --git a/tests/requirements-test.txt b/tests/requirements-test.txt new file mode 100644 index 0000000..cd7793a --- /dev/null +++ b/tests/requirements-test.txt @@ -0,0 +1,4 @@ +pytest>=7.0 +pytest-cov>=4.0 +pytest-mock>=3.10 +responses>=0.23 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/helpers/__init__.py b/tests/unit/helpers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/helpers/test_base.py b/tests/unit/helpers/test_base.py new file mode 100644 index 0000000..de1c40c --- /dev/null +++ b/tests/unit/helpers/test_base.py @@ -0,0 +1,60 @@ +"""Tests for ConsoleLog and Base classes — standalone, no API overlap.""" + +from unittest.mock import patch +from io import StringIO + +import pytest + +from pyzm.helpers.Base import Base, ConsoleLog + + +@pytest.mark.unit +class TestConsoleLog: + + def test_debug_respects_level(self): + """Debug messages above the set level are suppressed.""" + log = ConsoleLog() + log.set_level(2) + + with patch("sys.stdout", new_callable=StringIO) as mock_out: + log.Debug(1, "visible") + log.Debug(2, "also visible") + log.Debug(3, "suppressed") + + output = mock_out.getvalue() + assert "visible" in output + assert "also visible" in output + assert "suppressed" not in output + + def test_debug_level_getter(self): + """get_level returns current level.""" + log = ConsoleLog() + log.set_level(3) + assert log.get_level() == 3 + + def test_fatal_calls_exit(self, no_exit): + """Fatal prints message then calls exit(-1).""" + log = ConsoleLog() + + with patch("sys.stdout", new_callable=StringIO): + log.Fatal("critical error") + + no_exit.assert_called_once_with(-1) + + def test_panic_calls_exit(self, no_exit): + """Panic prints message then calls exit(-2).""" + log = ConsoleLog() + + with patch("sys.stdout", new_callable=StringIO): + log.Panic("panic error") + + no_exit.assert_called_once_with(-2) + + +@pytest.mark.unit +class TestBase: + + def test_base_instantiation(self): + """Base class can be instantiated.""" + b = Base() + assert b is not None diff --git a/tests/unit/helpers/test_events.py b/tests/unit/helpers/test_events.py new file mode 100644 index 0000000..a3cc551 --- /dev/null +++ b/tests/unit/helpers/test_events.py @@ -0,0 +1,131 @@ +"""Edge-case tests for Events helper — URL filter building logic.""" + +from unittest.mock import MagicMock, patch + +import pytest +import responses + +from pyzm.helpers.Events import Events + + +def _make_api_mock(events_data=None): + """Create a mock API that returns canned event data.""" + if events_data is None: + events_data = { + "events": [], + "pagination": {"count": 0, "current": 0, "nextPage": False}, + } + api = MagicMock() + api.api_url = "https://zm.example.com/zm/api" + api._make_request.return_value = events_data + return api + + +@pytest.mark.unit +class TestEventFilters: + + def test_filter_url_with_monitor_id(self): + """MonitorId filter appears in the URL.""" + api = _make_api_mock() + + Events(api=api, options={"mid": 5}) + + call_url = api._make_request.call_args[1]["url"] + assert "/MonitorId =:5" in call_url + + def test_filter_url_with_object_only(self): + """object_only appends REGEXP filter.""" + api = _make_api_mock() + + Events(api=api, options={"object_only": True}) + + call_url = api._make_request.call_args[1]["url"] + assert "/Notes REGEXP:detected:" in call_url + + def test_filter_url_with_event_id(self): + """event_id filter appears in URL.""" + api = _make_api_mock() + + Events(api=api, options={"event_id": "42"}) + + call_url = api._make_request.call_args[1]["url"] + assert "/Id=:42" in call_url + + def test_filter_url_with_alarmed_frames(self): + """min/max alarmed frames filters appear in URL.""" + api = _make_api_mock() + + Events( + api=api, + options={"min_alarmed_frames": 10, "max_alarmed_frames": 100}, + ) + + call_url = api._make_request.call_args[1]["url"] + assert "/AlarmFrames >=:10" in call_url + assert "/AlarmFrames <=:100" in call_url + + @patch("pyzm.helpers.Events.dateparser") + def test_filter_url_with_time_range(self, mock_dateparser): + """'from' with 'X to Y' splits into start/end time filters.""" + from datetime import datetime + + start = datetime(2024, 1, 15, 9, 0, 0) + end = datetime(2024, 1, 15, 10, 0, 0) + mock_dateparser.parse.side_effect = [start, end] + + api = _make_api_mock() + + Events(api=api, options={"from": "1 hour ago to now"}) + + call_url = api._make_request.call_args[1]["url"] + assert "/StartTime >=:2024-01-15 09:00:00" in call_url + assert "/StartTime <=:2024-01-15 10:00:00" in call_url + + +@pytest.mark.unit +class TestPagination: + + def test_pagination_stops_at_max_events(self): + """Loop terminates when currevents >= max_events.""" + page1 = { + "events": [ + {"Event": {"Id": "1", "Name": "E1", "MonitorId": "1", + "Cause": "", "Notes": "", "StartTime": "", + "EndTime": "", "Length": "1", "Frames": "1", + "AlarmFrames": "1", "TotScore": "1", + "AvgScore": "1", "MaxScore": "1", + "DefaultVideo": None, "FileSystemPath": ""}}, + ], + "pagination": { + "count": 5, + "current": 1, + "nextPage": True, + "page": 1, + }, + } + page2 = { + "events": [ + {"Event": {"Id": "2", "Name": "E2", "MonitorId": "1", + "Cause": "", "Notes": "", "StartTime": "", + "EndTime": "", "Length": "1", "Frames": "1", + "AlarmFrames": "1", "TotScore": "1", + "AvgScore": "1", "MaxScore": "1", + "DefaultVideo": None, "FileSystemPath": ""}}, + ], + "pagination": { + "count": 5, + "current": 1, + "nextPage": True, + "page": 2, + }, + } + + api = MagicMock() + api.api_url = "https://zm.example.com/zm/api" + api._make_request.side_effect = [page1, page2] + + events = Events(api=api, options={"max_events": 2}) + + # Should have fetched 2 pages (1 event each, limit=2) + assert len(events.list()) == 2 + assert api._make_request.call_count == 2 diff --git a/tests/unit/helpers/test_monitor.py b/tests/unit/helpers/test_monitor.py new file mode 100644 index 0000000..275828a --- /dev/null +++ b/tests/unit/helpers/test_monitor.py @@ -0,0 +1,118 @@ +"""Edge-case tests for Monitor helper — only logic NOT covered by API tests.""" + +from unittest.mock import MagicMock + +import pytest + +from pyzm.helpers.Monitor import Monitor + + +def _make_monitor(overrides=None): + """Create a Monitor instance with a mock API and test data.""" + data = { + "Monitor": { + "Id": "5", + "Name": "Test Camera", + "Function": "Modect", + "Enabled": "1", + "Type": "Ffmpeg", + "Width": "1920", + "Height": "1080", + } + } + if overrides: + data["Monitor"].update(overrides) + api = MagicMock() + api.api_url = "https://zm.example.com/zm/api" + return Monitor(monitor=data, api=api) + + +@pytest.mark.unit +class TestMonitorEnabled: + + def test_monitor_enabled_false(self): + """Enabled == '0' returns False.""" + mon = _make_monitor({"Enabled": "0"}) + assert mon.enabled() is False + + def test_monitor_enabled_true(self): + """Enabled == '1' returns True.""" + mon = _make_monitor({"Enabled": "1"}) + assert mon.enabled() is True + + +@pytest.mark.unit +class TestSetParameter: + + def test_set_parameter_builds_payload(self): + """Correct Monitor[Function] etc. payload keys are built.""" + mon = _make_monitor() + + mon.set_parameter(options={ + "function": "Record", + "name": "New Name", + "enabled": True, + }) + + mon.api._make_request.assert_called_once() + call_kwargs = mon.api._make_request.call_args + payload = call_kwargs[1]["payload"] + assert payload["Monitor[Function]"] == "Record" + assert payload["Monitor[Name]"] == "New Name" + assert payload["Monitor[Enabled]"] == "1" + + def test_set_parameter_disabled(self): + """enabled=False sends '0'.""" + mon = _make_monitor() + + mon.set_parameter(options={"enabled": False}) + + call_kwargs = mon.api._make_request.call_args + payload = call_kwargs[1]["payload"] + assert payload["Monitor[Enabled]"] == "0" + + def test_set_parameter_with_raw(self): + """Raw parameter passthrough works.""" + mon = _make_monitor() + + mon.set_parameter(options={ + "raw": {"Monitor[Colours]": "4", "Monitor[Method]": "simple"}, + }) + + call_kwargs = mon.api._make_request.call_args + payload = call_kwargs[1]["payload"] + assert payload["Monitor[Colours]"] == "4" + assert payload["Monitor[Method]"] == "simple" + + def test_set_parameter_empty_noop(self): + """No payload = no API call.""" + mon = _make_monitor() + + result = mon.set_parameter(options={}) + + mon.api._make_request.assert_not_called() + assert result is None + + +@pytest.mark.unit +class TestArmDisarm: + + def test_arm_url(self): + """arm() calls correct alarm command URL.""" + mon = _make_monitor() + + mon.arm() + + call_kwargs = mon.api._make_request.call_args + url = call_kwargs[1]["url"] + assert "/monitors/alarm/id:5/command:on.json" in url + + def test_disarm_url(self): + """disarm() calls correct alarm command URL.""" + mon = _make_monitor() + + mon.disarm() + + call_kwargs = mon.api._make_request.call_args + url = call_kwargs[1]["url"] + assert "/monitors/alarm/id:5/command:off.json" in url diff --git a/tests/unit/helpers/test_state.py b/tests/unit/helpers/test_state.py new file mode 100644 index 0000000..8327169 --- /dev/null +++ b/tests/unit/helpers/test_state.py @@ -0,0 +1,35 @@ +"""Edge-case tests for State helper.""" + +from unittest.mock import MagicMock + +import pytest + +from pyzm.helpers.State import State + + +@pytest.mark.unit +class TestStateEdgeCases: + + def test_active_false(self): + """IsActive != '1' returns False.""" + state = State( + state={"State": {"Id": "2", "Name": "away", "IsActive": "0", "Definition": "All active"}}, + api=MagicMock(), + ) + assert state.active() is False + + def test_definition_none(self): + """Empty definition returns None.""" + state = State( + state={"State": {"Id": "3", "Name": "home", "IsActive": "0", "Definition": ""}}, + api=MagicMock(), + ) + assert state.definition() is None + + def test_definition_with_value(self): + """Non-empty definition returns the string.""" + state = State( + state={"State": {"Id": "1", "Name": "default", "IsActive": "1", "Definition": "ZM default"}}, + api=MagicMock(), + ) + assert state.definition() == "ZM default" diff --git a/tests/unit/helpers/test_states.py b/tests/unit/helpers/test_states.py new file mode 100644 index 0000000..5304b98 --- /dev/null +++ b/tests/unit/helpers/test_states.py @@ -0,0 +1,68 @@ +"""Edge-case tests for States helper — search/find logic.""" + +from unittest.mock import MagicMock + +import pytest + +from pyzm.helpers.State import State +from pyzm.helpers.States import States + + +def _make_states(): + """Build a States-like object with pre-populated state list (no API call).""" + states_obj = States.__new__(States) + states_obj.api = MagicMock() + states_obj.states = [ + State( + state={"State": {"Id": "1", "Name": "default", "IsActive": "1", "Definition": ""}}, + api=states_obj.api, + ), + State( + state={"State": {"Id": "2", "Name": "Away", "IsActive": "0", "Definition": "All active"}}, + api=states_obj.api, + ), + State( + state={"State": {"Id": "3", "Name": "Home", "IsActive": "0", "Definition": ""}}, + api=states_obj.api, + ), + ] + return states_obj + + +@pytest.mark.unit +class TestStatesFind: + + def test_find_by_id(self): + """Exact id match returns correct State.""" + states = _make_states() + + result = states.find(id=2) + + assert result is not None + assert result.id() == 2 + assert result.name() == "Away" + + def test_find_by_name_case_insensitive(self): + """name.lower() comparison works regardless of case.""" + states = _make_states() + + result = states.find(name="away") + assert result is not None + assert result.name() == "Away" + + result2 = states.find(name="AWAY") + assert result2 is not None + assert result2.name() == "Away" + + def test_find_no_match_returns_none(self): + """Non-existent id/name returns None.""" + states = _make_states() + + assert states.find(id=999) is None + assert states.find(name="nonexistent") is None + + def test_find_no_args_returns_none(self): + """find() with no arguments returns None.""" + states = _make_states() + + assert states.find() is None diff --git a/tests/unit/test_api_auth.py b/tests/unit/test_api_auth.py new file mode 100644 index 0000000..0ac043a --- /dev/null +++ b/tests/unit/test_api_auth.py @@ -0,0 +1,294 @@ +"""Tests for ZMApi authentication flows.""" + +import datetime +from unittest.mock import patch + +import pytest +import responses + +from pyzm.api import ZMApi + + +@pytest.mark.unit +class TestLoginJWT: + """JWT token-based login (ZM API >= 2.0).""" + + @responses.activate + def test_login_jwt_success(self, zm_options, login_success_response): + """POST login with user/pass stores access and refresh tokens.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + api = ZMApi(options=zm_options.copy()) + + assert api.authenticated is True + assert api.auth_enabled is True + assert api.access_token == "test_access_token_abc123" + assert api.refresh_token == "test_refresh_token_xyz789" + assert api.access_token_expires == 3600 + assert api.refresh_token_expires == 86400 + assert api.access_token_datetime is not None + assert api.refresh_token_datetime is not None + assert api.api_version == "2.0" + assert api.zm_version == "1.36.32" + + @responses.activate + def test_login_jwt_stores_token_expiry_datetimes( + self, zm_options, login_success_response + ): + """Token expiry datetimes are in the future.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + now = datetime.datetime.now() + api = ZMApi(options=zm_options.copy()) + + assert api.access_token_datetime > now + assert api.refresh_token_datetime > now + + +@pytest.mark.unit +class TestLoginLegacy: + """Legacy credentials-based login (ZM API < 2.0).""" + + @responses.activate + def test_login_legacy_success(self, login_legacy_response): + """Legacy login stores credentials string.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_legacy_response, + status=200, + ) + + options = { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "user": "admin", + "password": "secret", + "disable_ssl_cert_check": True, + } + api = ZMApi(options=options) + + assert api.authenticated is True + assert api.legacy_credentials == "auth=abc123hash" + assert api.api_version == "1.0" + + @responses.activate + def test_login_legacy_append_password(self): + """When append_password is '1', password is appended to credentials.""" + legacy_response = { + "version": "1.32.3", + "apiversion": "1.0", + "credentials": "auth=abc123hash", + "append_password": "1", + } + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=legacy_response, + status=200, + ) + + options = { + "apiurl": "https://zm.example.com/zm/api", + "portalurl": "https://zm.example.com/zm", + "user": "admin", + "password": "mypass", + "disable_ssl_cert_check": True, + } + api = ZMApi(options=options) + + assert api.legacy_credentials == "auth=abc123hashmypass" + + +@pytest.mark.unit +class TestLoginNoAuth: + """Login without authentication credentials.""" + + @responses.activate + def test_login_no_auth(self, zm_options_no_auth, version_response): + """No user/password triggers GET to version endpoint.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/getVersion.json", + json=version_response, + status=200, + ) + + api = ZMApi(options=zm_options_no_auth.copy()) + + assert api.authenticated is True + assert api.auth_enabled is False + assert api.api_version == "2.0" + assert api.zm_version == "1.36.32" + + +@pytest.mark.unit +class TestLoginFailure: + """Login error handling.""" + + @responses.activate + def test_login_failure_raises(self, zm_options): + """401 from login raises HTTPError.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json={"error": "Unauthorized"}, + status=401, + ) + + import requests + + with pytest.raises(requests.exceptions.HTTPError): + ZMApi(options=zm_options.copy()) + + @responses.activate + def test_login_token_fallback(self, zm_options_token, login_success_response): + """Token auth 401 falls back to user/password.""" + # First call with token returns 401 + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json={"error": "Token revoked"}, + status=401, + ) + # Second call with user/pass succeeds + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + api = ZMApi(options=zm_options_token.copy()) + + assert api.authenticated is True + assert api.access_token == "test_access_token_abc123" + # Two POST requests made + assert len(responses.calls) == 2 + + +@pytest.mark.unit +class TestPortalURL: + """Portal URL guessing logic.""" + + @responses.activate + def test_portal_url_guessed(self, login_success_response): + """When only apiurl is provided, portal URL is derived by stripping /api.""" + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + options = { + "apiurl": "https://zm.example.com/zm/api", + "user": "admin", + "password": "secret", + "disable_ssl_cert_check": True, + } + api = ZMApi(options=options) + + assert api.portal_url == "https://zm.example.com/zm" + + +@pytest.mark.unit +class TestTokenRefresh: + """Token refresh and relogin logic.""" + + def test_token_no_refresh_when_fresh(self, zm_api): + """Token with >5min remaining skips refresh.""" + # Set expiry far in the future + zm_api.access_token_datetime = datetime.datetime.now() + datetime.timedelta( + hours=1 + ) + zm_api.refresh_token_datetime = datetime.datetime.now() + datetime.timedelta( + hours=24 + ) + + # Should not raise or trigger relogin + zm_api._refresh_tokens_if_needed() + # Token unchanged + assert zm_api.access_token == "test_access_token_abc123" + + @responses.activate + def test_token_refresh_when_near_expiry( + self, zm_api, login_success_response + ): + """_refresh_tokens_if_needed triggers relogin when <5min remaining.""" + # Set access token to expire in 2 minutes + zm_api.access_token_datetime = datetime.datetime.now() + datetime.timedelta( + minutes=2 + ) + # Refresh token still valid + zm_api.refresh_token_datetime = datetime.datetime.now() + datetime.timedelta( + hours=24 + ) + + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + zm_api._refresh_tokens_if_needed() + assert len(responses.calls) == 1 + + def test_token_refresh_skipped_when_no_expiry(self, zm_api): + """No expiry set means no refresh attempt.""" + zm_api.access_token_expires = None + zm_api.refresh_token_expires = None + + # Should return without doing anything + zm_api._refresh_tokens_if_needed() + + @responses.activate + def test_relogin_uses_refresh_token(self, zm_api, login_success_response): + """_relogin prefers refresh token when it has >5min remaining.""" + zm_api.refresh_token_datetime = datetime.datetime.now() + datetime.timedelta( + hours=24 + ) + + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + zm_api._relogin() + + # The options should have been updated to use refresh token + assert zm_api.options["token"] == "test_refresh_token_xyz789" + + @responses.activate + def test_relogin_uses_credentials_when_refresh_expired( + self, zm_api, login_success_response + ): + """_relogin falls back to user/pass when refresh token near expiry.""" + zm_api.refresh_token_datetime = datetime.datetime.now() + datetime.timedelta( + seconds=30 + ) + + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + + zm_api._relogin() + + # Token should have been cleared to force user/pass login + assert zm_api.options.get("token") is None diff --git a/tests/unit/test_api_methods.py b/tests/unit/test_api_methods.py new file mode 100644 index 0000000..a88a0ea --- /dev/null +++ b/tests/unit/test_api_methods.py @@ -0,0 +1,388 @@ +"""Tests for ZMApi public methods (monitors, events, states, configs, etc.). + +These are the primary coverage layer -- helper objects get tested as part +of the API contract. If an API test already asserts a helper behavior, +there is no separate helper test for it. +""" + +import pytest +import responses + +from pyzm.helpers.Configs import Configs +from pyzm.helpers.Event import Event +from pyzm.helpers.Events import Events +from pyzm.helpers.Monitor import Monitor +from pyzm.helpers.Monitors import Monitors +from pyzm.helpers.State import State +from pyzm.helpers.States import States + + +# --------------------------------------------------------------------------- +# Monitors +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestMonitors: + + @responses.activate + def test_monitors_returns_monitors_object( + self, zm_api, monitors_response + ): + """monitors() returns a Monitors instance with correct Monitor objects.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + ) + + result = zm_api.monitors() + + assert isinstance(result, Monitors) + monitors = result.list() + assert len(monitors) == 2 + + m1 = monitors[0] + assert isinstance(m1, Monitor) + assert m1.id() == 1 + assert m1.name() == "Front Door" + assert m1.function() == "Modect" + assert m1.enabled() is True + assert m1.type() == "Ffmpeg" + assert m1.dimensions() == {"width": 1920, "height": 1080} + + m2 = monitors[1] + assert m2.id() == 2 + assert m2.name() == "Backyard" + assert m2.enabled() is False + + @responses.activate + def test_monitors_cached(self, zm_api, monitors_response): + """Second call returns cached Monitors without new API request.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + ) + + first = zm_api.monitors() + second = zm_api.monitors() + + assert first is second + assert len(responses.calls) == 1 + + @responses.activate + def test_monitors_force_reload(self, zm_api, monitors_response): + """force_reload=True re-fetches from API.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + ) + responses.add( + responses.GET, + "https://zm.example.com/zm/api/monitors.json", + json=monitors_response, + status=200, + ) + + zm_api.monitors() + zm_api.monitors(options={"force_reload": True}) + + assert len(responses.calls) == 2 + + +# --------------------------------------------------------------------------- +# Events +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestEvents: + + @responses.activate + def test_events_returns_events_object(self, zm_api, events_response): + """events() returns an Events instance with correct Event objects.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/events/index.json", + json=events_response, + status=200, + ) + + result = zm_api.events() + + assert isinstance(result, Events) + events = result.list() + assert len(events) == 2 + assert result.count() == 2 + + e1 = events[0] + assert isinstance(e1, Event) + assert e1.id() == 100 + assert e1.name() == "Event 100" + assert e1.monitor_id() == 1 + assert e1.cause() == "Motion" + assert e1.notes() == "detected:person" + assert e1.duration() == 60.5 + assert e1.total_frames() == 150 + assert e1.alarmed_frames() == 45 + assert e1.score() == {"total": 500.0, "average": 11.1, "max": 85.0} + assert e1.video_file() == "100-video.mp4" + assert e1.fspath() == "/var/lib/zoneminder/events/1/2024-01-15/100" + + e2 = events[1] + assert e2.id() == 99 + assert e2.video_file() is None + + +# --------------------------------------------------------------------------- +# States +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestStates: + + @responses.activate + def test_states_returns_states_object(self, zm_api, states_response): + """states() returns a States instance with correct State objects.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/states.json", + json=states_response, + status=200, + ) + + result = zm_api.states() + + assert isinstance(result, States) + states = result.list() + assert len(states) == 3 + + s1 = states[0] + assert isinstance(s1, State) + assert s1.id() == 1 + assert s1.name() == "default" + assert s1.active() is True + assert s1.definition() == "ZM default state" + + s2 = states[1] + assert s2.id() == 2 + assert s2.name() == "away" + assert s2.active() is False + + +# --------------------------------------------------------------------------- +# Configs +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestConfigs: + + @responses.activate + def test_configs_returns_configs_object(self, zm_api, configs_response): + """configs() returns a Configs instance.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/configs.json", + json=configs_response, + status=200, + ) + + result = zm_api.configs() + + assert isinstance(result, Configs) + assert len(result.list()) == 3 + + @responses.activate + def test_configs_cached(self, zm_api, configs_response): + """Second call returns cached Configs.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/configs.json", + json=configs_response, + status=200, + ) + + first = zm_api.configs() + second = zm_api.configs() + + assert first is second + assert len(responses.calls) == 1 + + @responses.activate + def test_configs_find_no_args_returns_none(self, zm_api, configs_response): + """find() with no args returns None without iterating.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/configs.json", + json=configs_response, + status=200, + ) + configs = zm_api.configs() + assert configs.find() is None + + @responses.activate + def test_configs_set_name_none_returns_none(self, zm_api, configs_response): + """set(name=None) returns None (early guard).""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/configs.json", + json=configs_response, + status=200, + ) + configs = zm_api.configs() + assert configs.set(name=None, val="anything") is None + + @responses.activate + def test_configs_set_val_none_returns_none(self, zm_api, configs_response): + """set(val=None) returns None (early guard).""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/configs.json", + json=configs_response, + status=200, + ) + configs = zm_api.configs() + assert configs.set(name="ZM_AUTH_TYPE", val=None) is None + + +# --------------------------------------------------------------------------- +# Version +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestVersion: + + def test_version_returns_dict(self, zm_api): + """version() returns status, api_version, zm_version.""" + result = zm_api.version() + + assert result["status"] == "ok" + assert result["api_version"] == "2.0" + assert result["zm_version"] == "1.36.32" + + @responses.activate + def test_version_unauthenticated(self, zm_options_no_auth, version_response): + """version() returns error when not authenticated.""" + from pyzm.api import ZMApi + + # The no-auth login path sets authenticated=True too, so we need + # to force the unauthenticated state by making login fail + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/getVersion.json", + json=version_response, + status=200, + ) + api = ZMApi(options=zm_options_no_auth.copy()) + # Force unauthenticated state + api.authenticated = False + + result = api.version() + assert result["status"] == "error" + assert "reason" in result + + +# --------------------------------------------------------------------------- +# set_state / restart / stop / start +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestStateControl: + + @responses.activate + def test_set_state(self, zm_api): + """set_state calls correct URL.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/states/change/mystate.json", + json={"result": "ok"}, + status=200, + ) + + result = zm_api.set_state("mystate") + + assert result == {"result": "ok"} + assert "states/change/mystate.json" in responses.calls[0].request.url + + @responses.activate + def test_restart_calls_set_state(self, zm_api): + """restart() calls set_state('restart').""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/states/change/restart.json", + json={"result": "ok"}, + status=200, + ) + + zm_api.restart() + + assert "states/change/restart.json" in responses.calls[0].request.url + + @responses.activate + def test_stop_calls_set_state(self, zm_api): + """stop() calls set_state('stop').""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/states/change/stop.json", + json={"result": "ok"}, + status=200, + ) + + zm_api.stop() + + assert "states/change/stop.json" in responses.calls[0].request.url + + @responses.activate + def test_start_calls_set_state(self, zm_api): + """start() calls set_state('start').""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/states/change/start.json", + json={"result": "ok"}, + status=200, + ) + + zm_api.start() + + assert "states/change/start.json" in responses.calls[0].request.url + + def test_set_state_none_returns_none(self, zm_api): + """set_state(None) returns None without making a request.""" + result = zm_api.set_state(None) + assert result is None + + +# --------------------------------------------------------------------------- +# get_auth +# --------------------------------------------------------------------------- + +@pytest.mark.unit +class TestGetAuth: + + def test_get_auth_jwt(self, zm_api): + """get_auth returns 'token=XXX' for JWT auth.""" + result = zm_api.get_auth() + assert result == "token=test_access_token_abc123" + + def test_get_auth_legacy(self, zm_api_legacy): + """get_auth returns legacy credentials string.""" + result = zm_api_legacy.get_auth() + assert result == "auth=abc123hash" + + @responses.activate + def test_get_auth_disabled(self, zm_options_no_auth, version_response): + """get_auth returns empty string when auth disabled.""" + from pyzm.api import ZMApi + + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/getVersion.json", + json=version_response, + status=200, + ) + + api = ZMApi(options=zm_options_no_auth.copy()) + assert api.get_auth() == "" diff --git a/tests/unit/test_api_request.py b/tests/unit/test_api_request.py new file mode 100644 index 0000000..24ff8e3 --- /dev/null +++ b/tests/unit/test_api_request.py @@ -0,0 +1,289 @@ +"""Tests for ZMApi._make_request — retry logic, error handling, content types.""" + +import datetime +from unittest.mock import MagicMock, patch + +import pytest +import requests +import responses + +from pyzm.api import ZMApi + + +@pytest.mark.unit +class TestMakeRequestHTTPMethods: + """Verify each HTTP method dispatches correctly.""" + + @responses.activate + def test_make_request_get(self, zm_api): + responses.add( + responses.GET, + "https://zm.example.com/zm/api/test.json", + json={"ok": True}, + status=200, + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json" + ) + + assert result == {"ok": True} + assert responses.calls[0].request.method == "GET" + + @responses.activate + def test_make_request_post(self, zm_api): + responses.add( + responses.POST, + "https://zm.example.com/zm/api/test.json", + json={"created": True}, + status=200, + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json", + payload={"key": "val"}, + type="post", + ) + + assert result == {"created": True} + assert responses.calls[0].request.method == "POST" + + @responses.activate + def test_make_request_put(self, zm_api): + responses.add( + responses.PUT, + "https://zm.example.com/zm/api/test.json", + json={"updated": True}, + status=200, + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json", + payload={"key": "val"}, + type="put", + ) + + assert result == {"updated": True} + assert responses.calls[0].request.method == "PUT" + + @responses.activate + def test_make_request_delete(self, zm_api): + responses.add( + responses.DELETE, + "https://zm.example.com/zm/api/test.json", + body=b"", + status=200, + content_type="application/json", + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json", + type="delete", + ) + + assert result is None + assert responses.calls[0].request.method == "DELETE" + + def test_make_request_invalid_type_returns_none(self, zm_api): + """Invalid type logs error but ValueError is swallowed by the handler.""" + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json", + type="patch", + ) + assert result is None + + +@pytest.mark.unit +class TestMakeRequestReauth: + """401 and RELOGIN retry behavior.""" + + @responses.activate + def test_make_request_401_triggers_relogin_retry( + self, zm_api, login_success_response + ): + """401 triggers _relogin then retries the request once.""" + # Ensure refresh token is valid for relogin + zm_api.refresh_token_datetime = ( + datetime.datetime.now() + datetime.timedelta(hours=24) + ) + + # First request: 401 + responses.add( + responses.GET, + "https://zm.example.com/zm/api/test.json", + json={"error": "Unauthorized"}, + status=401, + ) + # Relogin + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + # Retry: success + responses.add( + responses.GET, + "https://zm.example.com/zm/api/test.json", + json={"ok": True}, + status=200, + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json" + ) + + assert result == {"ok": True} + # 3 calls: original GET, login POST, retry GET + assert len(responses.calls) == 3 + + @responses.activate + def test_make_request_401_no_retry_when_reauth_false(self, zm_api): + """401 with reauth=False does not retry.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/test.json", + json={"error": "Unauthorized"}, + status=401, + ) + + # Should not raise (the code silently returns None for unhandled 401 w/ reauth=False) + result = zm_api._make_request( + url="https://zm.example.com/zm/api/test.json", + reauth=False, + ) + + assert len(responses.calls) == 1 + + +@pytest.mark.unit +class TestMakeRequestContentTypes: + """Response content-type handling.""" + + @responses.activate + def test_make_request_json_response(self, zm_api): + """application/json response is parsed to dict.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/data.json", + json={"data": [1, 2, 3]}, + status=200, + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/data.json" + ) + + assert result == {"data": [1, 2, 3]} + + @responses.activate + def test_make_request_image_response(self, zm_api): + """image/* content-type returns the raw response object.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/image.jpg", + body=b"\xff\xd8\xff\xe0", + status=200, + content_type="image/jpeg", + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/image.jpg" + ) + + # Returns response object, not parsed JSON + assert hasattr(result, "status_code") + assert result.status_code == 200 + + @responses.activate + def test_make_request_404_raises_bad_image(self, zm_api): + """404 raises ValueError('BAD_IMAGE').""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/image.jpg", + json={"error": "Not found"}, + status=404, + ) + + with pytest.raises(ValueError, match="BAD_IMAGE"): + zm_api._make_request( + url="https://zm.example.com/zm/api/image.jpg" + ) + + @responses.activate + def test_make_request_relogin_value_error( + self, zm_api, login_success_response + ): + """Non-JSON, non-image response with content triggers RELOGIN retry.""" + zm_api.refresh_token_datetime = ( + datetime.datetime.now() + datetime.timedelta(hours=24) + ) + + # First: returns HTML (not JSON, not image, has content) + responses.add( + responses.GET, + "https://zm.example.com/zm/api/resource.json", + body=b"Login required", + status=200, + content_type="text/html", + headers={"content-length": "27"}, + ) + # Relogin + responses.add( + responses.POST, + "https://zm.example.com/zm/api/host/login.json", + json=login_success_response, + status=200, + ) + # Retry: proper JSON + responses.add( + responses.GET, + "https://zm.example.com/zm/api/resource.json", + json={"data": "ok"}, + status=200, + ) + + result = zm_api._make_request( + url="https://zm.example.com/zm/api/resource.json" + ) + + assert result == {"data": "ok"} + assert len(responses.calls) == 3 + + +@pytest.mark.unit +class TestMakeRequestTokenInjection: + """Verify tokens/credentials are injected into requests.""" + + @responses.activate + def test_jwt_token_added_to_query(self, zm_api): + """JWT access token is added as query parameter.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/test.json", + json={"ok": True}, + status=200, + ) + + zm_api._make_request(url="https://zm.example.com/zm/api/test.json") + + request_url = responses.calls[0].request.url + assert "token=test_access_token_abc123" in request_url + + @responses.activate + def test_legacy_credentials_appended_to_url(self, zm_api_legacy): + """Legacy credentials are appended to URL.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/test.json", + json={"ok": True}, + status=200, + match_querystring=False, + ) + + zm_api_legacy._make_request( + url="https://zm.example.com/zm/api/test.json" + ) + + request_url = responses.calls[0].request.url + assert "auth=abc123hash" in request_url From b09135f52f522e4f3fc07e413c8d32808162352c Mon Sep 17 00:00:00 2001 From: Nic Boet Date: Fri, 13 Feb 2026 14:52:12 -0600 Subject: [PATCH 2/4] Add E2E test suite for testing pyzm against live ZoneMinder 97 tests (81 readonly, 16 write) that validate pyzm against a real ZM instance, catching response structure drift, type coercion mismatches, and flash() redirect handling that unit tests with mocked HTTP miss. Tests skip automatically when ZM_API_URL env vars are unset. Co-Authored-By: Claude Opus 4.6 --- pyproject.toml | 3 + tests/README.md | 59 ++++++++ tests/e2e/__init__.py | 0 tests/e2e/conftest.py | 174 +++++++++++++++++++++++ tests/e2e/test_e2e_auth.py | 92 ++++++++++++ tests/e2e/test_e2e_configs.py | 104 ++++++++++++++ tests/e2e/test_e2e_edge_cases.py | 104 ++++++++++++++ tests/e2e/test_e2e_events.py | 220 ++++++++++++++++++++++++++++ tests/e2e/test_e2e_monitors.py | 237 +++++++++++++++++++++++++++++++ tests/e2e/test_e2e_states.py | 158 +++++++++++++++++++++ 10 files changed, 1151 insertions(+) create mode 100644 tests/e2e/__init__.py create mode 100644 tests/e2e/conftest.py create mode 100644 tests/e2e/test_e2e_auth.py create mode 100644 tests/e2e/test_e2e_configs.py create mode 100644 tests/e2e/test_e2e_edge_cases.py create mode 100644 tests/e2e/test_e2e_events.py create mode 100644 tests/e2e/test_e2e_monitors.py create mode 100644 tests/e2e/test_e2e_states.py diff --git a/pyproject.toml b/pyproject.toml index a782fca..5adaf49 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,6 +4,9 @@ markers = [ "unit: Unit tests (fast, no external deps)", "integration: Integration tests (multi-step workflows)", "slow: Slow tests", + "e2e: All end-to-end tests (require live ZoneMinder)", + "e2e_readonly: E2E tests that only read data (safe to run anytime)", + "e2e_write: E2E tests that create/modify/delete data (opt-in via ZM_E2E_WRITE=1)", ] addopts = "-v --strict-markers --tb=short" diff --git a/tests/README.md b/tests/README.md index d201a3c..9a36279 100644 --- a/tests/README.md +++ b/tests/README.md @@ -124,3 +124,62 @@ pytest tests/unit/test_api_auth.py -v - `pyzm.ZMEventNotification` — Requires WebSocket - `pyzm.helpers.Media` — Requires cv2/numpy - `pyzm.helpers.utils` — `draw_bbox` requires cv2; `Timer`/`read_config`/`template_fill` are simple utilities + +--- + +## E2E Tests (End-to-End) + +### Overview + +E2E tests run against a **live ZoneMinder instance**. They catch bugs that unit tests miss because unit tests mock HTTP responses with hand-crafted JSON fixtures. + +| Category | Example | Why unit tests miss it | +|----------|---------|----------------------| +| Response structure drift | ZM returns `StartDateTime` but pyzm expects `StartTime` | Fixture uses whatever pyzm expects | +| flash() instead of JSON | `MonitorsController.delete()` returns HTML redirect | Fixture mocks clean JSON | +| Filter URL building bugs | pyzm builds wrong filter path -> wrong results | Mock returns whatever you tell it | +| Type coercion mismatches | ZM returns `"1"` (string), pyzm assumes `int` | Fixture can use any type | +| Auth flow against real server | JWT format, token refresh timing, credential format | Mock always returns 200 | + +### E2E Environment Setup + +| Variable | Required | Description | +|----------|----------|-------------| +| `ZM_API_URL` | Yes | Full API URL, e.g. `https://zm.local/zm/api` | +| `ZM_USER` | Yes | ZoneMinder username | +| `ZM_PASSWORD` | Yes | ZoneMinder password | +| `ZM_E2E_WRITE` | No | Set to `1` to enable write-tier tests | + +If env vars are unset, all E2E tests are skipped automatically. + +### Running E2E Tests + +```bash +# Readonly only (safe, no data changes) +ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret \ + pytest tests/e2e/ -m e2e_readonly + +# Write tests (creates/modifies/deletes with cleanup) +ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret ZM_E2E_WRITE=1 \ + pytest tests/e2e/ -m e2e_write + +# All E2E +ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret ZM_E2E_WRITE=1 \ + pytest tests/e2e/ + +# Collect-only (verify discovery without a live instance) +pytest tests/e2e/ -v --co -m e2e_readonly +``` + +### E2E Tiers + +- **Readonly** (`e2e_readonly`): list, find, get, filter operations. Safe to run repeatedly. +- **Write** (`e2e_write`): create, modify, delete operations. Require `ZM_E2E_WRITE=1`. All write tests clean up: + - Monitors prefixed `pyzm_e2e_test_` are deleted in teardown + - Config values saved before mutation and restored in teardown + - States recorded and restored after switching + +### Known pyzm Bugs Documented as E2E Tests + +- `Configs.find(name="nonexistent")` raises `TypeError` at `Configs.py:64` (no null check on `match`) +- Monitor/event delete may return `None` (flash redirect) instead of JSON diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..149a1e7 --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,174 @@ +"""E2E test fixtures for testing pyzm against a live ZoneMinder instance. + +Requires environment variables: + ZM_API_URL - Full API URL (e.g. https://zm.local/zm/api) + ZM_USER - ZoneMinder username + ZM_PASSWORD - ZoneMinder password + ZM_E2E_WRITE - Set to "1" to enable write-tier tests (optional) +""" + +import os + +import pytest + + +# --------------------------------------------------------------------------- +# Override parent autouse fixtures +# --------------------------------------------------------------------------- +# The parent tests/conftest.py defines autouse fixtures `suppress_logger` and +# `no_exit` that mock the logger and builtins.exit(). E2E tests need the real +# logger and real exit() so we override with no-op fixtures at this scope. +# pytest resolves fixtures from the closest conftest first. + +@pytest.fixture(autouse=True) +def suppress_logger(): + """No-op override: let real logger run during E2E tests.""" + yield None + + +@pytest.fixture(autouse=True) +def no_exit(): + """No-op override: let real exit() work during E2E tests.""" + yield None + + +# --------------------------------------------------------------------------- +# Environment & skip helpers +# --------------------------------------------------------------------------- + +def _get_zm_env(): + """Read ZM connection info from environment. Returns dict or None.""" + api_url = os.environ.get("ZM_API_URL") + user = os.environ.get("ZM_USER") + password = os.environ.get("ZM_PASSWORD") + if not all([api_url, user, password]): + return None + return { + "apiurl": api_url, + "user": user, + "password": password, + "disable_ssl_cert_check": True, + } + + +def _write_enabled(): + return os.environ.get("ZM_E2E_WRITE") == "1" + + +# --------------------------------------------------------------------------- +# Session-scoped fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="session") +def zm_options_live(): + """ZM options dict from env vars. Skips entire session if unset.""" + env = _get_zm_env() + if env is None: + pytest.skip("ZM_API_URL / ZM_USER / ZM_PASSWORD not set") + return env + + +@pytest.fixture(scope="session") +def zm_api_live(zm_options_live): + """Single authenticated ZMApi for the entire test session. + + Token refresh is handled internally by ZMApi._refresh_tokens_if_needed(). + """ + from pyzm.api import ZMApi + + api = ZMApi(options=zm_options_live.copy()) + assert api.authenticated is True, "E2E: login to live ZM failed" + return api + + +# --------------------------------------------------------------------------- +# Function-scoped fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def zm_api_fresh(zm_options_live): + """Fresh ZMApi login per test (for auth-specific tests).""" + from pyzm.api import ZMApi + + return ZMApi(options=zm_options_live.copy()) + + +@pytest.fixture +def e2e_monitor_factory(zm_api_live): + """Factory that creates monitors with auto-cleanup. + + Usage: + mon = e2e_monitor_factory(name="test cam", function="Monitor") + # ... test ... + # teardown deletes all created monitors + """ + created = [] + + def _create(**kwargs): + name = kwargs.pop("name", "pyzm_e2e_test_monitor") + if not name.startswith("pyzm_e2e_test_"): + name = "pyzm_e2e_test_" + name + opts = { + "name": name, + "function": kwargs.pop("function", "Monitor"), + "enabled": kwargs.pop("enabled", False), + "width": kwargs.pop("width", 640), + "height": kwargs.pop("height", 480), + "raw": kwargs.pop("raw", {}), + } + opts.update(kwargs) + result = zm_api_live.monitors({"force_reload": True}).add(options=opts) + # Reload monitors to find the newly created one + monitors = zm_api_live.monitors({"force_reload": True}) + mon = monitors.find(name=name) + if mon is not None: + created.append(mon) + return mon, result + + yield _create + + # Teardown: delete all monitors we created + for mon in created: + try: + mon.delete() + except Exception: + pass + + +@pytest.fixture +def e2e_config_restorer(zm_api_live): + """Records original config value and restores it in teardown. + + Usage: + e2e_config_restorer("ZM_LANG_DEFAULT") + configs.set(name="ZM_LANG_DEFAULT", val="de_DE") + # ... test ... + # teardown restores original value + """ + saved = [] + + def _save(config_name): + configs = zm_api_live.configs({"force_reload": True}) + original = configs.find(name=config_name) + saved.append((config_name, original["value"])) + + yield _save + + # Teardown: restore all saved configs + for config_name, original_value in saved: + try: + configs = zm_api_live.configs({"force_reload": True}) + configs.set(name=config_name, val=original_value) + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Skip helpers as fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def requires_write(): + """Skip test if ZM_E2E_WRITE is not set.""" + if not _write_enabled(): + pytest.skip("ZM_E2E_WRITE not set to 1") diff --git a/tests/e2e/test_e2e_auth.py b/tests/e2e/test_e2e_auth.py new file mode 100644 index 0000000..e5b60c6 --- /dev/null +++ b/tests/e2e/test_e2e_auth.py @@ -0,0 +1,92 @@ +"""E2E tests for ZMApi authentication and basic server info. + +All tests are readonly — they only query the server. +""" + +import re + +import pytest + +pytestmark = [pytest.mark.e2e, pytest.mark.e2e_readonly] + + +class TestLogin: + """Verify login succeeds and sets expected attributes.""" + + def test_login_succeeds(self, zm_api_live): + assert zm_api_live.authenticated is True + + def test_login_sets_api_version(self, zm_api_live): + v = zm_api_live.api_version + assert v is not None + assert isinstance(v, str) + # API version should be a dotted number like "2.0" + assert re.match(r"^\d+\.\d+", v), f"Unexpected api_version format: {v}" + + def test_login_sets_zm_version(self, zm_api_live): + v = zm_api_live.zm_version + assert v is not None + assert isinstance(v, str) + # ZM version should look like "1.36.33" or similar + assert re.match(r"^\d+\.\d+", v), f"Unexpected zm_version format: {v}" + + +class TestVersion: + """Verify version() returns proper structure.""" + + def test_version_returns_ok_status(self, zm_api_live): + result = zm_api_live.version() + assert result["status"] == "ok" + + def test_version_has_expected_keys(self, zm_api_live): + result = zm_api_live.version() + assert "api_version" in result + assert "zm_version" in result + assert isinstance(result["api_version"], str) + assert isinstance(result["zm_version"], str) + + +class TestTimezone: + """Verify tz() returns Area/Location format.""" + + def test_tz_returns_valid_timezone(self, zm_api_live): + tz = zm_api_live.tz() + assert tz is not None + assert isinstance(tz, str) + # Timezone should be in Area/Location format (e.g. "America/New_York") + assert "/" in tz, f"Timezone not in Area/Location format: {tz}" + + +class TestGetAuth: + """Verify get_auth() returns a usable auth string.""" + + def test_get_auth_returns_string(self, zm_api_live): + auth = zm_api_live.get_auth() + assert isinstance(auth, str) + # Should be either "token=..." or "auth=..." depending on API version + assert auth.startswith("token=") or auth.startswith("auth="), \ + f"Unexpected auth format: {auth}" + + def test_get_auth_nonempty(self, zm_api_live): + auth = zm_api_live.get_auth() + assert len(auth) > 6, "Auth string too short to contain a real token" + + +class TestFreshLogin: + """Verify a fresh login works (function-scoped, not session-cached).""" + + def test_fresh_login_authenticates(self, zm_api_fresh): + assert zm_api_fresh.authenticated is True + assert zm_api_fresh.api_version is not None + + +class TestBadCredentials: + """Verify bad credentials raise an error.""" + + def test_bad_password_raises(self, zm_options_live): + from pyzm.api import ZMApi + + bad_opts = zm_options_live.copy() + bad_opts["password"] = "definitely_wrong_password_xyz" + with pytest.raises(Exception): + ZMApi(options=bad_opts) diff --git a/tests/e2e/test_e2e_configs.py b/tests/e2e/test_e2e_configs.py new file mode 100644 index 0000000..fea9140 --- /dev/null +++ b/tests/e2e/test_e2e_configs.py @@ -0,0 +1,104 @@ +"""E2E tests for Configs against a live ZoneMinder. + +Readonly tests verify list/find. Write tests modify config values (with restore). +""" + +import pytest + +from pyzm.helpers.Configs import Configs + +pytestmark = [pytest.mark.e2e] + + +# --------------------------------------------------------------------------- +# Readonly tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_readonly +class TestConfigList: + """Verify configs() returns a valid Configs collection.""" + + def test_returns_configs_instance(self, zm_api_live): + result = zm_api_live.configs() + assert isinstance(result, Configs) + + def test_list_returns_list(self, zm_api_live): + configs = zm_api_live.configs() + lst = configs.list() + assert isinstance(lst, list) + + def test_list_nonempty(self, zm_api_live): + configs = zm_api_live.configs() + assert len(configs.list()) > 0 + + def test_item_structure(self, zm_api_live): + """Each config item should be a dict with {'Config': {'Id', 'Name', 'Value'}}.""" + configs = zm_api_live.configs() + item = configs.list()[0] + assert isinstance(item, dict) + assert "Config" in item + inner = item["Config"] + assert "Id" in inner + assert "Name" in inner + assert "Value" in inner + + +@pytest.mark.e2e_readonly +class TestConfigFind: + """Verify Configs.find() search logic.""" + + def test_find_by_name(self, zm_api_live): + configs = zm_api_live.configs({"force_reload": True}) + result = configs.find(name="ZM_AUTH_TYPE") + assert result is not None + assert isinstance(result, dict) + assert isinstance(result["id"], int) + assert isinstance(result["name"], str) + assert isinstance(result["value"], str) + assert result["name"] == "ZM_AUTH_TYPE" + + def test_find_zm_auth_type(self, zm_api_live): + """ZM_AUTH_TYPE should exist on any ZM installation.""" + configs = zm_api_live.configs({"force_reload": True}) + result = configs.find(name="ZM_AUTH_TYPE") + assert result is not None + # Value should be "builtin" or "remote" + assert result["value"] in ("builtin", "remote"), \ + f"Unexpected ZM_AUTH_TYPE value: {result['value']}" + + def test_find_nonexistent_raises_typeerror(self, zm_api_live): + """Documents pyzm bug: Configs.find() at line 64 crashes when no match. + + Configs.find() doesn't check if `match` is None before accessing + match['Config']['Id']. This raises TypeError. + Unlike Monitors.find() and States.find() which return None. + """ + configs = zm_api_live.configs({"force_reload": True}) + with pytest.raises(TypeError): + configs.find(name="ZM_PYZM_E2E_NONEXISTENT_CONFIG_XYZ") + + +# --------------------------------------------------------------------------- +# Write tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_write +class TestConfigSet: + """Test modifying config values (with auto-restore).""" + + def test_set_config_value(self, zm_api_live, e2e_config_restorer, requires_write): + """Modify a safe config value and verify the change.""" + config_name = "ZM_LANG_DEFAULT" + e2e_config_restorer(config_name) + + configs = zm_api_live.configs({"force_reload": True}) + original = configs.find(name=config_name) + new_value = "de_DE" if original["value"] != "de_DE" else "en_GB" + + configs.set(name=config_name, val=new_value) + + # Reload and verify + configs = zm_api_live.configs({"force_reload": True}) + updated = configs.find(name=config_name) + assert updated["value"] == new_value + diff --git a/tests/e2e/test_e2e_edge_cases.py b/tests/e2e/test_e2e_edge_cases.py new file mode 100644 index 0000000..9d9c486 --- /dev/null +++ b/tests/e2e/test_e2e_edge_cases.py @@ -0,0 +1,104 @@ +"""E2E tests for ZM API edge cases, type coercion, and quirks. + +These tests validate behaviors that are easy to get wrong with mocked responses: +type coercion mismatches, pagination coherence, and non-JSON response handling. +""" + +import pytest + +pytestmark = [pytest.mark.e2e] + + +# --------------------------------------------------------------------------- +# Readonly tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_readonly +class TestEventsPagination: + """Verify events pagination count vs list length coherence.""" + + def test_count_gte_list_length(self, zm_api_live): + """count() returns total matching events; list() is capped by max_events.""" + events = zm_api_live.events({"max_events": 3}) + count = events.count() + length = len(events.list()) + assert isinstance(count, int) + assert isinstance(length, int) + # list length should not exceed requested max + assert length <= 3 + # count is total matching, should be >= what we got + assert count >= length + + +@pytest.mark.e2e_readonly +class TestTimezoneConsistency: + """Verify timezone is consistent across calls.""" + + def test_tz_same_across_calls(self, zm_api_live): + tz1 = zm_api_live.tz() + tz2 = zm_api_live.tz() + assert tz1 == tz2 + + +@pytest.mark.e2e_readonly +class TestVersionParseable: + """Verify version fields can be parsed as integers.""" + + def test_api_version_parts_are_ints(self, zm_api_live): + v = zm_api_live.version() + parts = v["api_version"].split(".") + for part in parts: + int(part) # Should not raise + + def test_zm_version_parts_are_ints(self, zm_api_live): + v = zm_api_live.version() + parts = v["zm_version"].split(".") + for part in parts: + int(part) # Should not raise + + +@pytest.mark.e2e_readonly +class TestMonitorIdTypeCoercion: + """Verify monitor ID type coercion: accessor always returns int.""" + + def test_round_trip_id(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + assert int(mon.get()["Id"]) == mon.id() + + +@pytest.mark.e2e_readonly +class TestEventScoreTypeCoercion: + """Verify event score type coercion matches raw data.""" + + def test_score_float_matches_raw(self, zm_api_live): + events = zm_api_live.events({"max_events": 1}) + lst = events.list() + if not lst: + pytest.skip("No events available") + ev = lst[0] + raw = ev.get() + score = ev.score() + assert score["total"] == float(raw["TotScore"]) + assert score["average"] == float(raw["AvgScore"]) + assert score["max"] == float(raw["MaxScore"]) + + +# --------------------------------------------------------------------------- +# Write tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_write +class TestArmDisarm: + """Test arm/disarm handles non-JSON responses gracefully.""" + + def test_arm_disarm(self, zm_api_live, e2e_monitor_factory, requires_write): + mon, _ = e2e_monitor_factory(name="arm_test", function="Monitor") + assert mon is not None + # arm() may return dict or None (ZM may return non-JSON) + result = mon.arm() + assert result is None or isinstance(result, dict) + # disarm() + result = mon.disarm() + assert result is None or isinstance(result, dict) + + diff --git a/tests/e2e/test_e2e_events.py b/tests/e2e/test_e2e_events.py new file mode 100644 index 0000000..1b00fb6 --- /dev/null +++ b/tests/e2e/test_e2e_events.py @@ -0,0 +1,220 @@ +"""E2E tests for Events and Event objects against a live ZoneMinder. + +Readonly tests verify list/filter/accessors. Write tests delete events. +""" + +import pytest + +from pyzm.helpers.Events import Events +from pyzm.helpers.Event import Event + +pytestmark = [pytest.mark.e2e] + + +# --------------------------------------------------------------------------- +# Readonly tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_readonly +class TestEventList: + """Verify events() returns a valid Events collection.""" + + def test_returns_events_instance(self, zm_api_live): + result = zm_api_live.events() + assert isinstance(result, Events) + + def test_list_returns_list(self, zm_api_live): + events = zm_api_live.events() + lst = events.list() + assert isinstance(lst, list) + + def test_items_are_event_objects(self, zm_api_live): + events = zm_api_live.events({"max_events": 5}) + for ev in events.list(): + assert isinstance(ev, Event) + + def test_count_returns_int(self, zm_api_live): + events = zm_api_live.events({"max_events": 5}) + count = events.count() + assert isinstance(count, int) + assert count >= 0 + + +@pytest.mark.e2e_readonly +class TestEventAccessors: + """Verify Event accessor methods return correct types.""" + + @pytest.fixture + def first_event(self, zm_api_live): + events = zm_api_live.events({"max_events": 5}) + lst = events.list() + if not lst: + pytest.skip("No events available on live ZM") + return lst[0] + + def test_id_is_int(self, first_event): + assert isinstance(first_event.id(), int) + assert first_event.id() > 0 + + def test_monitor_id_is_int(self, first_event): + assert isinstance(first_event.monitor_id(), int) + assert first_event.monitor_id() > 0 + + def test_duration_is_float(self, first_event): + assert isinstance(first_event.duration(), float) + assert first_event.duration() >= 0 + + def test_score_dict(self, first_event): + score = first_event.score() + assert isinstance(score, dict) + assert isinstance(score["total"], float) + assert isinstance(score["average"], float) + assert isinstance(score["max"], float) + assert score["average"] <= score["max"] or score["max"] == 0 + + def test_total_frames_is_int(self, first_event): + assert isinstance(first_event.total_frames(), int) + assert first_event.total_frames() >= 0 + + def test_alarmed_frames_is_int(self, first_event): + assert isinstance(first_event.alarmed_frames(), int) + assert first_event.alarmed_frames() >= 0 + assert first_event.alarmed_frames() <= first_event.total_frames() + + def test_name_is_str(self, first_event): + name = first_event.name() + # name can be None if the Event Name field is empty + assert name is None or isinstance(name, str) + + def test_cause_is_str(self, first_event): + cause = first_event.cause() + assert cause is None or isinstance(cause, str) + + +@pytest.mark.e2e_readonly +class TestEventRawGet: + """Verify Event.get() returns raw dict with expected ZM fields.""" + + def test_raw_dict_has_required_fields(self, zm_api_live): + events = zm_api_live.events({"max_events": 1}) + lst = events.list() + if not lst: + pytest.skip("No events available on live ZM") + raw = lst[0].get() + assert isinstance(raw, dict) + # String fields — Name/Cause are always str in ZM's JSON + for field in ["Name", "Cause"]: + assert field in raw, f"Missing field: {field}" + assert isinstance(raw[field], (str, type(None))), \ + f"Expected str or None for raw field {field}, got {type(raw[field])}" + # Numeric fields — ZM returns int/float, older versions may return str + for field in ["Id", "MonitorId", "Frames", "AlarmFrames", + "TotScore", "AvgScore", "MaxScore"]: + assert field in raw, f"Missing field: {field}" + assert isinstance(raw[field], (int, float, str)), \ + f"Expected int/float/str for raw field {field}, got {type(raw[field])}" + # Length (duration) — numeric + assert "Length" in raw, "Missing field: Length" + assert isinstance(raw["Length"], (int, float, str)), \ + f"Expected int/float/str for raw field Length, got {type(raw['Length'])}" + + +@pytest.mark.e2e_readonly +class TestEventFilters: + """Verify event filtering works against real ZM.""" + + def test_filter_by_max_events(self, zm_api_live): + events = zm_api_live.events({"max_events": 3}) + assert len(events.list()) <= 3 + + def test_filter_by_monitor_id(self, zm_api_live): + # Get a monitor that exists + monitors = zm_api_live.monitors() + if not monitors.list(): + pytest.skip("No monitors on live ZM") + mid = monitors.list()[0].id() + events = zm_api_live.events({"mid": mid, "max_events": 5}) + for ev in events.list(): + assert ev.monitor_id() == mid + + def test_filter_by_min_alarmed_frames(self, zm_api_live): + events = zm_api_live.events({"min_alarmed_frames": 1, "max_events": 5}) + for ev in events.list(): + assert ev.alarmed_frames() >= 1 + + def test_empty_result_for_nonexistent_monitor(self, zm_api_live): + events = zm_api_live.events({"mid": 999999, "max_events": 5}) + assert len(events.list()) == 0 + + +@pytest.mark.e2e_readonly +class TestMonitorEvents: + """Verify Monitor.events() convenience method.""" + + def test_monitor_events_returns_events(self, zm_api_live): + monitors = zm_api_live.monitors() + if not monitors.list(): + pytest.skip("No monitors on live ZM") + mon = monitors.list()[0] + events = mon.events({"max_events": 3}) + assert isinstance(events, Events) + for ev in events.list(): + assert ev.monitor_id() == mon.id() + + +@pytest.mark.e2e_readonly +class TestEventUrls: + """Verify image/video URL generation.""" + + def test_image_url_format(self, zm_api_live): + events = zm_api_live.events({"max_events": 1}) + lst = events.list() + if not lst: + pytest.skip("No events available on live ZM") + url = lst[0].get_image_url() + assert isinstance(url, str) + assert "index.php" in url + assert "eid=" in url + # Should contain auth token or credentials + assert "token=" in url or "auth=" in url + + def test_video_url_format(self, zm_api_live): + events = zm_api_live.events({"max_events": 1}) + lst = events.list() + if not lst: + pytest.skip("No events available on live ZM") + ev = lst[0] + url = ev.get_video_url() + # video_url can be None if no video file + if ev.video_file(): + assert isinstance(url, str) + assert "eid=" in url + + +# --------------------------------------------------------------------------- +# Write tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_write +class TestEventDelete: + """Test deleting an event.""" + + def test_delete_event(self, zm_api_live, requires_write): + """Delete the oldest event (least impactful). + + Note: ZM's EventsController.delete() may return flash redirect (None) + instead of JSON — this test verifies pyzm handles that gracefully. + """ + events = zm_api_live.events({ + "max_events": 1, + "sort": "StartTime", + "direction": "asc", + }) + lst = events.list() + if not lst: + pytest.skip("No events available to delete") + ev = lst[0] + eid = ev.id() + result = ev.delete() + # ZM may return None (flash redirect) or dict — both acceptable + assert result is None or isinstance(result, dict) diff --git a/tests/e2e/test_e2e_monitors.py b/tests/e2e/test_e2e_monitors.py new file mode 100644 index 0000000..77591b4 --- /dev/null +++ b/tests/e2e/test_e2e_monitors.py @@ -0,0 +1,237 @@ +"""E2E tests for Monitors and Monitor objects against a live ZoneMinder. + +Readonly tests verify list/find/accessors. Write tests create, modify, +and delete monitors (with auto-cleanup). +""" + +import pytest + +from pyzm.helpers.Monitors import Monitors +from pyzm.helpers.Monitor import Monitor + +# --------------------------------------------------------------------------- +# Readonly tests +# --------------------------------------------------------------------------- + +pytestmark = [pytest.mark.e2e] + + +@pytest.mark.e2e_readonly +class TestMonitorList: + """Verify monitors() returns a valid Monitors collection.""" + + def test_returns_monitors_instance(self, zm_api_live): + result = zm_api_live.monitors() + assert isinstance(result, Monitors) + + def test_list_returns_list(self, zm_api_live): + monitors = zm_api_live.monitors() + lst = monitors.list() + assert isinstance(lst, list) + + def test_list_nonempty(self, zm_api_live): + monitors = zm_api_live.monitors() + assert len(monitors.list()) > 0, "Expected at least one monitor on live ZM" + + def test_items_are_monitor_objects(self, zm_api_live): + monitors = zm_api_live.monitors() + for mon in monitors.list(): + assert isinstance(mon, Monitor) + + +@pytest.mark.e2e_readonly +class TestMonitorAccessors: + """Verify Monitor accessor methods return correct types.""" + + def test_id_is_int(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + assert isinstance(mon.id(), int) + assert mon.id() > 0 + + def test_name_is_str(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + assert isinstance(mon.name(), str) + assert len(mon.name()) > 0 + + def test_function_is_valid(self, zm_api_live): + valid_functions = { + "None", "Monitor", "Modect", "Record", + "Mocord", "Nodect", + } + mon = zm_api_live.monitors().list()[0] + func = mon.function() + assert isinstance(func, str) + assert func in valid_functions, f"Unexpected function: {func}" + + def test_enabled_is_bool(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + assert isinstance(mon.enabled(), bool) + + def test_dimensions_dict(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + dims = mon.dimensions() + assert isinstance(dims, dict) + assert isinstance(dims["width"], int) + assert isinstance(dims["height"], int) + assert dims["width"] > 0 + assert dims["height"] > 0 + + def test_type_is_str(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + assert isinstance(mon.type(), str) + + +@pytest.mark.e2e_readonly +class TestMonitorRawGet: + """Verify Monitor.get() returns raw dict with expected ZM fields.""" + + def test_raw_dict_has_required_fields(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + raw = mon.get() + assert isinstance(raw, dict) + # String fields — always str in ZM's JSON + for field in ["Name", "Function", "Type"]: + assert field in raw, f"Missing field: {field}" + assert isinstance(raw[field], str), \ + f"Expected str for raw field {field}, got {type(raw[field])}" + # Id — confirmed int on real ZM + assert isinstance(raw["Id"], int), \ + f"Expected int for raw field Id, got {type(raw['Id'])}" + # Numeric fields — ZM returns int, older versions may return str + for field in ["Enabled", "Width", "Height"]: + assert field in raw, f"Missing field: {field}" + assert isinstance(raw[field], (int, str)), \ + f"Expected int or str for raw field {field}, got {type(raw[field])}" + + +@pytest.mark.e2e_readonly +class TestMonitorFind: + """Verify Monitors.find() by id and name.""" + + def test_find_by_id(self, zm_api_live): + monitors = zm_api_live.monitors() + first = monitors.list()[0] + found = monitors.find(id=first.id()) + assert found is not None + assert found.id() == first.id() + + def test_find_by_name(self, zm_api_live): + monitors = zm_api_live.monitors() + first = monitors.list()[0] + found = monitors.find(name=first.name()) + assert found is not None + assert found.name() == first.name() + + def test_find_nonexistent_returns_none(self, zm_api_live): + monitors = zm_api_live.monitors() + found = monitors.find(name="pyzm_e2e_nonexistent_monitor_xyz") + assert found is None + + +@pytest.mark.e2e_readonly +class TestMonitorDaemonStatus: + """Verify daemon status returns a response.""" + + def test_status_returns_dict(self, zm_api_live): + mon = zm_api_live.monitors().list()[0] + result = mon.status() + # status() may return a dict or None depending on ZM configuration + assert result is None or isinstance(result, dict) + + +# --------------------------------------------------------------------------- +# Write tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_write +class TestMonitorAdd: + """Test creating a new monitor.""" + + def test_add_monitor(self, zm_api_live, e2e_monitor_factory): + mon, result = e2e_monitor_factory(name="add_test", function="Monitor") + assert mon is not None, "Failed to find newly created monitor" + assert mon.name() == "pyzm_e2e_test_add_test" + assert mon.function() == "Monitor" + + def test_add_monitor_with_dimensions(self, zm_api_live, e2e_monitor_factory): + mon, result = e2e_monitor_factory( + name="dims_test", + function="Monitor", + width=320, + height=240, + ) + assert mon is not None + dims = mon.dimensions() + assert dims["width"] == 320 + assert dims["height"] == 240 + + +@pytest.mark.e2e_write +class TestMonitorSetParameter: + """Test modifying monitor parameters.""" + + def test_set_function(self, zm_api_live, e2e_monitor_factory): + mon, _ = e2e_monitor_factory(name="setparam_func", function="Monitor") + assert mon is not None + mon.set_parameter({"function": "Modect"}) + # Reload to verify + monitors = zm_api_live.monitors({"force_reload": True}) + updated = monitors.find(name="pyzm_e2e_test_setparam_func") + assert updated is not None + assert updated.function() == "Modect" + + def test_set_name(self, zm_api_live, e2e_monitor_factory): + mon, _ = e2e_monitor_factory(name="setparam_name", function="Monitor") + assert mon is not None + mon.set_parameter({"name": "pyzm_e2e_test_setparam_renamed"}) + monitors = zm_api_live.monitors({"force_reload": True}) + updated = monitors.find(name="pyzm_e2e_test_setparam_renamed") + assert updated is not None + + def test_set_enabled(self, zm_api_live, e2e_monitor_factory): + mon, _ = e2e_monitor_factory(name="setparam_enabled", function="Monitor", enabled=False) + assert mon is not None + assert mon.enabled() is False + mon.set_parameter({"enabled": True}) + monitors = zm_api_live.monitors({"force_reload": True}) + updated = monitors.find(name="pyzm_e2e_test_setparam_enabled") + assert updated is not None + assert updated.enabled() is True + + def test_set_raw_parameter(self, zm_api_live, e2e_monitor_factory): + mon, _ = e2e_monitor_factory(name="setparam_raw", function="Monitor") + assert mon is not None + mon.set_parameter({"raw": {"Monitor[Colours]": "4"}}) + monitors = zm_api_live.monitors({"force_reload": True}) + updated = monitors.find(name="pyzm_e2e_test_setparam_raw") + assert updated is not None + assert updated.get()["Colours"] == "4" + + +@pytest.mark.e2e_write +class TestMonitorDelete: + """Test deleting a monitor.""" + + def test_delete_monitor(self, zm_api_live, e2e_monitor_factory): + mon, _ = e2e_monitor_factory(name="delete_test", function="Monitor") + assert mon is not None + mid = mon.id() + # Delete it manually (factory will also try in teardown, which is fine) + result = mon.delete() + # ZM may return None (flash redirect) or a dict — both are acceptable + assert result is None or isinstance(result, dict) + # Verify it's gone + monitors = zm_api_live.monitors({"force_reload": True}) + assert monitors.find(id=mid) is None + + +@pytest.mark.e2e_write +class TestMonitorAddFlashHandling: + """Test that add() handles flash() responses from ZM.""" + + def test_add_returns_response(self, zm_api_live, e2e_monitor_factory): + """ZM's MonitorsController.add() may return flash redirect or JSON.""" + _, result = e2e_monitor_factory(name="flash_test", function="Monitor") + # Result may be None (flash), dict (JSON), or response object + # The important thing is it doesn't crash + assert result is None or isinstance(result, dict) diff --git a/tests/e2e/test_e2e_states.py b/tests/e2e/test_e2e_states.py new file mode 100644 index 0000000..c85e60d --- /dev/null +++ b/tests/e2e/test_e2e_states.py @@ -0,0 +1,158 @@ +"""E2E tests for States and State objects against a live ZoneMinder. + +Readonly tests verify list/find/accessors. Write tests switch states. +""" + +import pytest + +from pyzm.helpers.States import States +from pyzm.helpers.State import State + +pytestmark = [pytest.mark.e2e] + + +# --------------------------------------------------------------------------- +# Readonly tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_readonly +class TestStateList: + """Verify states() returns a valid States collection.""" + + def test_returns_states_instance(self, zm_api_live): + result = zm_api_live.states() + assert isinstance(result, States) + + def test_list_returns_list(self, zm_api_live): + states = zm_api_live.states() + lst = states.list() + assert isinstance(lst, list) + + def test_list_nonempty(self, zm_api_live): + states = zm_api_live.states() + assert len(states.list()) > 0, "Expected at least one state on live ZM" + + def test_items_are_state_objects(self, zm_api_live): + states = zm_api_live.states() + for s in states.list(): + assert isinstance(s, State) + + +@pytest.mark.e2e_readonly +class TestStateAccessors: + """Verify State accessor methods return correct types.""" + + def test_id_is_int(self, zm_api_live): + state = zm_api_live.states().list()[0] + assert isinstance(state.id(), int) + assert state.id() > 0 + + def test_name_is_str(self, zm_api_live): + state = zm_api_live.states().list()[0] + assert isinstance(state.name(), str) + assert len(state.name()) > 0 + + def test_active_is_bool(self, zm_api_live): + state = zm_api_live.states().list()[0] + assert isinstance(state.active(), bool) + + def test_definition_is_str_or_none(self, zm_api_live): + state = zm_api_live.states().list()[0] + defn = state.definition() + assert defn is None or isinstance(defn, str) + + +@pytest.mark.e2e_readonly +class TestStateRawGet: + """Verify State.get() returns raw dict with expected ZM fields.""" + + def test_raw_dict_has_required_fields(self, zm_api_live): + state = zm_api_live.states().list()[0] + raw = state.get() + assert isinstance(raw, dict) + # String fields — Name is always str in ZM's JSON + assert "Name" in raw, "Missing field: Name" + assert isinstance(raw["Name"], str), \ + f"Expected str for raw field Name, got {type(raw['Name'])}" + # Numeric fields — ZM returns int, older versions may return str + for field in ["Id", "IsActive"]: + assert field in raw, f"Missing field: {field}" + assert isinstance(raw[field], (int, str)), \ + f"Expected int or str for raw field {field}, got {type(raw[field])}" + + +@pytest.mark.e2e_readonly +class TestStateFind: + """Verify States.find() by id and name.""" + + def test_find_by_id(self, zm_api_live): + states = zm_api_live.states() + first = states.list()[0] + found = states.find(id=first.id()) + assert found is not None + assert found.id() == first.id() + + def test_find_by_name(self, zm_api_live): + states = zm_api_live.states() + first = states.list()[0] + found = states.find(name=first.name()) + assert found is not None + assert found.name() == first.name() + + def test_find_nonexistent_returns_none(self, zm_api_live): + states = zm_api_live.states() + found = states.find(name="pyzm_e2e_nonexistent_state_xyz") + assert found is None + + +@pytest.mark.e2e_readonly +class TestStateActiveInvariant: + """Verify at most one state is active at a time.""" + + def test_at_most_one_active(self, zm_api_live): + states = zm_api_live.states() + active_count = sum(1 for s in states.list() if s.active()) + assert active_count <= 1, \ + f"Expected at most 1 active state, found {active_count}" + + +# --------------------------------------------------------------------------- +# Write tests +# --------------------------------------------------------------------------- + +@pytest.mark.e2e_write +class TestSetState: + """Test switching ZM state with restore.""" + + def test_set_state_and_restore(self, zm_api_live, requires_write): + """Record active state, switch to another, verify, restore.""" + states = zm_api_live.states() + state_list = states.list() + if len(state_list) < 2: + pytest.skip("Need at least 2 states to test switching") + + # Record the currently active state (if any) + active_state = None + for s in state_list: + if s.active(): + active_state = s + break + + # Pick a target state that's different from the active one + target = None + for s in state_list: + if not s.active(): + target = s + break + + if target is None: + pytest.skip("No inactive state to switch to") + + # Switch to target + result = zm_api_live.set_state(target.name()) + # set_state may return dict or None + assert result is None or isinstance(result, dict) + + # Restore original state if there was one + if active_state: + zm_api_live.set_state(active_state.name()) From 7f39903e33a2e296e920cd363e3409a2a2245282 Mon Sep 17 00:00:00 2001 From: Nic Boet Date: Fri, 13 Feb 2026 16:30:52 -0600 Subject: [PATCH 3/4] Fix pyzm bugs detected by E2E tests against live ZoneMinder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Running the E2E test suite against a real ZoneMinder instance exposed five bugs in pyzm that unit tests (with hand-crafted fixtures) never caught — exactly the kind of drift E2E tests are designed to surface. Monitors.add() missing type/device parameters: E2E test_add_monitor → ZM returned 500. Captured the response body and found: "Field 'Device' doesn't have a default value". The real API requires Monitor[Type] and Monitor[Device] but Monitors.add() had no support for either. Added both as first-class parameters. Monitors.add() enabled=False silently dropped: E2E test_set_enabled → monitor created with enabled=False was not actually disabled. The guard `if options.get('enabled')` is falsy for False, so Monitor[Enabled] was never sent. Changed to `if options.get('enabled') is not None`. Monitor.enabled() string vs int comparison: E2E test_set_enabled → after enabling a monitor, enabled() returned False. The real API returns Enabled as int 1, not string "1". The comparison `== '1'` fails for int. Fixed with str() coercion. _make_request raises BAD_IMAGE on successful PUT: E2E test_set_config_value → Configs.set() PUT returned 200 with Content-Length: 0. _make_request only handled empty bodies for DELETE, so PUT fell through to the BAD_IMAGE error path. Extended the empty-body handler to cover PUT. Also adjusted E2E test assertions for raw field types (Colours returns int from real API, not string as fixtures assumed). Co-Authored-By: Claude Opus 4.6 --- pyzm/api.py | 2 +- pyzm/helpers/Configs.py | 2 ++ pyzm/helpers/Monitor.py | 2 +- pyzm/helpers/Monitors.py | 11 ++++++++--- pyzm/helpers/State.py | 2 +- tests/e2e/conftest.py | 2 ++ tests/e2e/test_e2e_configs.py | 14 +++++--------- tests/e2e/test_e2e_monitors.py | 2 +- tests/unit/test_api_methods.py | 12 ++++++++++++ 9 files changed, 33 insertions(+), 16 deletions(-) diff --git a/pyzm/api.py b/pyzm/api.py index ab54385..210aa94 100644 --- a/pyzm/api.py +++ b/pyzm/api.py @@ -283,7 +283,7 @@ def _make_request(self, url=None, query={}, payload={}, type='get', reauth=True) return r.json() elif r.headers.get('content-type').startswith('image/'): return r - elif type=='delete': + elif type in ('delete', 'put'): return None else: # A non 0 byte response will usually mean its an image eid request that needs re-login diff --git a/pyzm/helpers/Configs.py b/pyzm/helpers/Configs.py index df62334..a4fbef7 100644 --- a/pyzm/helpers/Configs.py +++ b/pyzm/helpers/Configs.py @@ -61,6 +61,8 @@ def find(self, id=None, name=None): if name and config['Config']['Name'].lower() == name.lower(): match = config break + if match is None: + return None return { 'id': int(match['Config']['Id']), 'name': match['Config']['Name'], diff --git a/pyzm/helpers/Monitor.py b/pyzm/helpers/Monitor.py index d471493..421652f 100644 --- a/pyzm/helpers/Monitor.py +++ b/pyzm/helpers/Monitor.py @@ -31,7 +31,7 @@ def enabled(self): Returns: bool: Enabled or not """ - return self.monitor['Monitor']['Enabled'] == '1' + return str(self.monitor['Monitor']['Enabled']) == '1' def function(self): """returns monitor function diff --git a/pyzm/helpers/Monitors.py b/pyzm/helpers/Monitors.py index 10a3c96..9edae93 100644 --- a/pyzm/helpers/Monitors.py +++ b/pyzm/helpers/Monitors.py @@ -40,6 +40,8 @@ def add(self, options={}): { 'function': string # function of monitor 'name': string # name of monitor + 'type': string # capture type (Ffmpeg, Remote, Local, etc.) + 'device': string # device path (e.g. /dev/video0) 'enabled': boolean 'protocol': string 'host': string @@ -64,9 +66,12 @@ def add(self, options={}): payload['Monitor[Function]'] = options.get('function') if options.get('name'): payload['Monitor[Name]'] = options.get('name') - if options.get('enabled'): - enabled = '1' if options.get('enabled') else '0' - payload['Monitor[Enabled]'] = enabled + if options.get('type'): + payload['Monitor[Type]'] = options.get('type') + if options.get('device') is not None: + payload['Monitor[Device]'] = options.get('device') + if options.get('enabled') is not None: + payload['Monitor[Enabled]'] = '1' if options.get('enabled') else '0' if options.get('protocol'): payload['Monitor[Protocol]'] = options.get('protocol') if options.get('host'): diff --git a/pyzm/helpers/State.py b/pyzm/helpers/State.py index 24f3533..682844a 100644 --- a/pyzm/helpers/State.py +++ b/pyzm/helpers/State.py @@ -32,7 +32,7 @@ def active(self): Returns: bool: True if active """ - return self.state['State']['IsActive'] == '1' + return str(self.state['State']['IsActive']) == '1' def definition(self): """Returns the description text of this state diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 149a1e7..45f7ec4 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -110,6 +110,8 @@ def _create(**kwargs): name = "pyzm_e2e_test_" + name opts = { "name": name, + "type": kwargs.pop("type", "Ffmpeg"), + "device": kwargs.pop("device", ""), "function": kwargs.pop("function", "Monitor"), "enabled": kwargs.pop("enabled", False), "width": kwargs.pop("width", 640), diff --git a/tests/e2e/test_e2e_configs.py b/tests/e2e/test_e2e_configs.py index fea9140..72e57b9 100644 --- a/tests/e2e/test_e2e_configs.py +++ b/tests/e2e/test_e2e_configs.py @@ -66,16 +66,12 @@ def test_find_zm_auth_type(self, zm_api_live): assert result["value"] in ("builtin", "remote"), \ f"Unexpected ZM_AUTH_TYPE value: {result['value']}" - def test_find_nonexistent_raises_typeerror(self, zm_api_live): - """Documents pyzm bug: Configs.find() at line 64 crashes when no match. - - Configs.find() doesn't check if `match` is None before accessing - match['Config']['Id']. This raises TypeError. - Unlike Monitors.find() and States.find() which return None. - """ + def test_find_nonexistent_returns_none(self, zm_api_live): + """Configs.find() returns None when no match — consistent with + Monitors.find() and States.find().""" configs = zm_api_live.configs({"force_reload": True}) - with pytest.raises(TypeError): - configs.find(name="ZM_PYZM_E2E_NONEXISTENT_CONFIG_XYZ") + result = configs.find(name="ZM_PYZM_E2E_NONEXISTENT_CONFIG_XYZ") + assert result is None # --------------------------------------------------------------------------- diff --git a/tests/e2e/test_e2e_monitors.py b/tests/e2e/test_e2e_monitors.py index 77591b4..9535864 100644 --- a/tests/e2e/test_e2e_monitors.py +++ b/tests/e2e/test_e2e_monitors.py @@ -205,7 +205,7 @@ def test_set_raw_parameter(self, zm_api_live, e2e_monitor_factory): monitors = zm_api_live.monitors({"force_reload": True}) updated = monitors.find(name="pyzm_e2e_test_setparam_raw") assert updated is not None - assert updated.get()["Colours"] == "4" + assert str(updated.get()["Colours"]) == "4" @pytest.mark.e2e_write diff --git a/tests/unit/test_api_methods.py b/tests/unit/test_api_methods.py index a88a0ea..f343803 100644 --- a/tests/unit/test_api_methods.py +++ b/tests/unit/test_api_methods.py @@ -223,6 +223,18 @@ def test_configs_find_no_args_returns_none(self, zm_api, configs_response): configs = zm_api.configs() assert configs.find() is None + @responses.activate + def test_configs_find_nonexistent_returns_none(self, zm_api, configs_response): + """find() returns None when no config matches.""" + responses.add( + responses.GET, + "https://zm.example.com/zm/api/configs.json", + json=configs_response, + status=200, + ) + configs = zm_api.configs() + assert configs.find(name="ZM_NONEXISTENT_XYZ") is None + @responses.activate def test_configs_set_name_none_returns_none(self, zm_api, configs_response): """set(name=None) returns None (early guard).""" From 0ebf4cf3b74a3d3c972247b489a33e9ab825e7e0 Mon Sep 17 00:00:00 2001 From: Nic Boet Date: Fri, 13 Feb 2026 17:25:36 -0600 Subject: [PATCH 4/4] Remove redundant helper tests, update README with E2E design intent Removed 3 helper tests that duplicated assertions already covered by API-level tests (per project rule). Rewrote README to lead with the core philosophy: the real ZM API is the source of truth. Co-Authored-By: Claude Opus 4.6 --- tests/README.md | 175 ++++++++++++++++------------- tests/unit/helpers/test_monitor.py | 14 --- tests/unit/helpers/test_state.py | 7 -- 3 files changed, 99 insertions(+), 97 deletions(-) diff --git a/tests/README.md b/tests/README.md index 9a36279..0fb0e7d 100644 --- a/tests/README.md +++ b/tests/README.md @@ -1,20 +1,42 @@ # pyzm Testing Strategy -## Philosophy +## Core Design Intent -pyzm testing follows a **public API first** approach: +**The real ZoneMinder API is the source of truth.** Hand-crafted JSON fixtures can diverge from actual server responses in type coercion, field names, response structure, and error formats. E2E tests against a live ZoneMinder instance are the authoritative validation layer. -- **Public API tests** (`tests/unit/test_api_*.py`) test `ZMApi` methods as consumers use them. All HTTP is mocked via `responses`. These are the **primary coverage layer** — since `zm_api.monitors()` creates `Monitors` → `Monitor` objects from API data, the happy path for helper accessors is already covered here. -- **Helper edge-case tests** (`tests/unit/helpers/`) cover ONLY edge cases, pure logic, and error paths not reachable through the public API. Examples: `States.find()` case-insensitive search, `Monitor.set_parameter()` payload construction. -- **Integration tests** (`tests/integration/`) chain multiple API calls in realistic workflows. +The test suite is structured in four tiers, each with a specific role: -**Rule: if an API test already asserts a helper behavior, don't write a separate helper test for it.** +1. **E2E tests** (`tests/e2e/`) — the gold standard. Run against a **live ZoneMinder instance** with real HTTP, real auth, real data. Every assertion is validated against actual server behavior. +2. **Unit API tests** (`tests/unit/test_api_*.py`) — the primary offline coverage layer. Test `ZMApi` public methods as consumers use them. HTTP is mocked via `responses`. Since `zm_api.monitors()` creates `Monitors` → `Monitor` objects from fixture data, the happy path for helper accessors is already covered here. +3. **Helper edge-case tests** (`tests/unit/helpers/`) — cover ONLY edge cases, pure logic, and error paths not reachable through the public API. Examples: `States.find()` case-insensitive search, `Monitor.set_parameter()` payload construction, `State.definition()` returning None for empty strings. +4. **Integration tests** (`tests/integration/`) — chain multiple mocked API calls in realistic workflows. + +### Rules + +- **If an API test already asserts a helper behavior, don't write a separate helper test for it.** Duplicate assertions across tiers create maintenance burden without adding confidence. +- **E2E tests intentionally re-validate behaviors covered by unit tests.** This is by design — E2E catches real-world divergence that fixtures miss. Overlap between E2E and unit tiers is expected and valuable. +- **JSON fixtures must mirror real ZM API payload structure**, including using strings for numeric fields (ZM returns `"1"` not `1` for many fields). E2E tests verify this assumption holds. + +## Why E2E Tests Exist + +Unit tests mock HTTP responses with hand-crafted JSON. This means unit tests **cannot catch** an entire class of bugs: + +| Category | Example | Why unit tests miss it | +|----------|---------|----------------------| +| Response structure drift | ZM returns `StartDateTime` but pyzm expects `StartTime` | Fixture uses whatever pyzm expects | +| flash() instead of JSON | `MonitorsController.delete()` returns HTML redirect | Fixture mocks clean JSON | +| Filter URL building bugs | pyzm builds wrong filter path → wrong results | Mock returns whatever you tell it | +| Type coercion mismatches | ZM returns `int` but fixture uses `str` (or vice versa) | Fixture can use any type | +| Auth flow against real server | JWT format, token refresh timing, credential format | Mock always returns 200 | +| Real-world data volume | Pagination with thousands of events | Mock returns static fixture | + +E2E tests are the only way to confirm pyzm works against the software it claims to wrap. ## Directory Structure ``` tests/ -├── conftest.py # Shared fixtures +├── conftest.py # Shared fixtures (autouse logger/exit patches) ├── fixtures/ │ └── responses/ # JSON response fixtures mirroring ZM API payloads │ ├── login_success.json @@ -33,28 +55,49 @@ tests/ │ ├── test_monitor.py # set_parameter payload, arm/disarm URLs │ ├── test_events.py # URL filter building, pagination │ ├── test_state.py # active() false, definition() None -│ ├── test_states.py # find() search logic +│ ├── test_states.py # find() search logic (case-insensitive, by id) │ └── test_base.py # ConsoleLog level filtering, exit calls -└── integration/ - └── test_api_workflow.py # Full login -> monitors -> events -> states +├── integration/ +│ └── test_api_workflow.py # Full login → monitors → events → states +└── e2e/ + ├── conftest.py # Live ZM fixtures, cleanup factories + ├── test_e2e_auth.py # Login, version, timezone, get_auth, bad creds + ├── test_e2e_monitors.py # List/find/accessors, add/modify/delete + ├── test_e2e_events.py # List/filter/accessors, URLs, delete + ├── test_e2e_states.py # List/find/active invariant, set state + ├── test_e2e_configs.py # List/find, set with restore + └── test_e2e_edge_cases.py # Pagination coherence, type coercion, arm/disarm ``` ## Fixture Patterns ### JSON Response Fixtures -Response fixtures in `tests/fixtures/responses/` mirror actual ZM API payloads. They are loaded by helper functions in `conftest.py` and provided as pytest fixtures. +Response fixtures in `tests/fixtures/responses/` mirror actual ZM API payloads. All numeric values are strings (e.g. `"Id": "1"`, `"Enabled": "1"`, `"Width": "1920"`) because that is what ZoneMinder's API returns. This keeps fixtures honest — if pyzm's type coercion breaks, the unit tests catch it. ### Key Shared Fixtures -| Fixture | Purpose | -|---|---| -| `zm_options` | Standard config dict for JWT auth | -| `zm_options_no_auth` | Config without credentials | -| `zm_api` | Pre-authenticated `ZMApi` with JWT login mocked | -| `zm_api_legacy` | Pre-authenticated `ZMApi` with legacy credentials | -| `suppress_logger` | (autouse) Patches `g.logger` to silent mock | -| `no_exit` | (autouse) Patches `builtins.exit` | +| Fixture | Scope | Purpose | +|---|---|---| +| `zm_options` | function | Standard config dict for JWT auth | +| `zm_options_no_auth` | function | Config without credentials | +| `zm_api` | function | Pre-authenticated `ZMApi` with JWT login mocked | +| `zm_api_legacy` | function | Pre-authenticated `ZMApi` with legacy credentials | +| `suppress_logger` | function (autouse) | Patches `g.logger` to silent mock | +| `no_exit` | function (autouse) | Patches `builtins.exit` | + +### E2E Fixtures + +| Fixture | Scope | Purpose | +|---|---|---| +| `zm_options_live` | session | Options dict from env vars (skips if unset) | +| `zm_api_live` | session | Single authenticated `ZMApi` for all E2E tests | +| `zm_api_fresh` | function | Fresh login per test (auth-specific tests) | +| `e2e_monitor_factory` | function | Creates monitors with auto-cleanup in teardown | +| `e2e_config_restorer` | function | Saves config value, restores in teardown | +| `requires_write` | function | Skips if `ZM_E2E_WRITE != "1"` | + +The E2E conftest overrides `suppress_logger` and `no_exit` with no-ops so E2E tests use the real logger and real `exit()`. ### Test Isolation @@ -84,8 +127,8 @@ pyzm uses a global logger at `pyzm.helpers.globals.logger`. The `suppress_logger ## Running Tests ```bash -# All tests -pytest tests/ -v +# All unit + integration tests (no live ZM needed) +pytest tests/unit/ tests/integration/ -v # With coverage pytest tests/ -v --cov=pyzm --cov-report=term-missing @@ -98,8 +141,42 @@ pytest tests/integration/ -v -m integration # Specific test file pytest tests/unit/test_api_auth.py -v + +# E2E readonly only (safe, no data changes) +ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret \ + pytest tests/e2e/ -m e2e_readonly + +# E2E write tests (creates/modifies/deletes with cleanup) +ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret ZM_E2E_WRITE=1 \ + pytest tests/e2e/ -m e2e_write + +# All E2E +ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret ZM_E2E_WRITE=1 \ + pytest tests/e2e/ + +# Collect-only (verify discovery without a live instance) +pytest tests/e2e/ -v --co -m e2e_readonly ``` +## E2E Environment Setup + +| Variable | Required | Description | +|----------|----------|-------------| +| `ZM_API_URL` | Yes | Full API URL, e.g. `https://zm.local/zm/api` | +| `ZM_USER` | Yes | ZoneMinder username | +| `ZM_PASSWORD` | Yes | ZoneMinder password | +| `ZM_E2E_WRITE` | No | Set to `1` to enable write-tier tests | + +If env vars are unset, all E2E tests are skipped automatically. + +### E2E Tiers + +- **Readonly** (`e2e_readonly`): list, find, get, filter operations. Safe to run repeatedly. +- **Write** (`e2e_write`): create, modify, delete operations. Require `ZM_E2E_WRITE=1`. All write tests clean up: + - Monitors prefixed `pyzm_e2e_test_` are deleted in teardown + - Config values saved before mutation and restored in teardown + - States recorded and restored after switching + ## Coverage Targets - Core modules (`api.py`, helpers): 80%+ @@ -125,61 +202,7 @@ pytest tests/unit/test_api_auth.py -v - `pyzm.helpers.Media` — Requires cv2/numpy - `pyzm.helpers.utils` — `draw_bbox` requires cv2; `Timer`/`read_config`/`template_fill` are simple utilities ---- - -## E2E Tests (End-to-End) - -### Overview - -E2E tests run against a **live ZoneMinder instance**. They catch bugs that unit tests miss because unit tests mock HTTP responses with hand-crafted JSON fixtures. - -| Category | Example | Why unit tests miss it | -|----------|---------|----------------------| -| Response structure drift | ZM returns `StartDateTime` but pyzm expects `StartTime` | Fixture uses whatever pyzm expects | -| flash() instead of JSON | `MonitorsController.delete()` returns HTML redirect | Fixture mocks clean JSON | -| Filter URL building bugs | pyzm builds wrong filter path -> wrong results | Mock returns whatever you tell it | -| Type coercion mismatches | ZM returns `"1"` (string), pyzm assumes `int` | Fixture can use any type | -| Auth flow against real server | JWT format, token refresh timing, credential format | Mock always returns 200 | - -### E2E Environment Setup - -| Variable | Required | Description | -|----------|----------|-------------| -| `ZM_API_URL` | Yes | Full API URL, e.g. `https://zm.local/zm/api` | -| `ZM_USER` | Yes | ZoneMinder username | -| `ZM_PASSWORD` | Yes | ZoneMinder password | -| `ZM_E2E_WRITE` | No | Set to `1` to enable write-tier tests | - -If env vars are unset, all E2E tests are skipped automatically. - -### Running E2E Tests - -```bash -# Readonly only (safe, no data changes) -ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret \ - pytest tests/e2e/ -m e2e_readonly - -# Write tests (creates/modifies/deletes with cleanup) -ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret ZM_E2E_WRITE=1 \ - pytest tests/e2e/ -m e2e_write - -# All E2E -ZM_API_URL=https://zm.local/zm/api ZM_USER=admin ZM_PASSWORD=secret ZM_E2E_WRITE=1 \ - pytest tests/e2e/ - -# Collect-only (verify discovery without a live instance) -pytest tests/e2e/ -v --co -m e2e_readonly -``` - -### E2E Tiers - -- **Readonly** (`e2e_readonly`): list, find, get, filter operations. Safe to run repeatedly. -- **Write** (`e2e_write`): create, modify, delete operations. Require `ZM_E2E_WRITE=1`. All write tests clean up: - - Monitors prefixed `pyzm_e2e_test_` are deleted in teardown - - Config values saved before mutation and restored in teardown - - States recorded and restored after switching - -### Known pyzm Bugs Documented as E2E Tests +## Known pyzm Bugs Documented as E2E Tests - `Configs.find(name="nonexistent")` raises `TypeError` at `Configs.py:64` (no null check on `match`) - Monitor/event delete may return `None` (flash redirect) instead of JSON diff --git a/tests/unit/helpers/test_monitor.py b/tests/unit/helpers/test_monitor.py index 275828a..cabc5d1 100644 --- a/tests/unit/helpers/test_monitor.py +++ b/tests/unit/helpers/test_monitor.py @@ -27,20 +27,6 @@ def _make_monitor(overrides=None): return Monitor(monitor=data, api=api) -@pytest.mark.unit -class TestMonitorEnabled: - - def test_monitor_enabled_false(self): - """Enabled == '0' returns False.""" - mon = _make_monitor({"Enabled": "0"}) - assert mon.enabled() is False - - def test_monitor_enabled_true(self): - """Enabled == '1' returns True.""" - mon = _make_monitor({"Enabled": "1"}) - assert mon.enabled() is True - - @pytest.mark.unit class TestSetParameter: diff --git a/tests/unit/helpers/test_state.py b/tests/unit/helpers/test_state.py index 8327169..28d3c62 100644 --- a/tests/unit/helpers/test_state.py +++ b/tests/unit/helpers/test_state.py @@ -26,10 +26,3 @@ def test_definition_none(self): ) assert state.definition() is None - def test_definition_with_value(self): - """Non-empty definition returns the string.""" - state = State( - state={"State": {"Id": "1", "Name": "default", "IsActive": "1", "Definition": "ZM default"}}, - api=MagicMock(), - ) - assert state.definition() == "ZM default"