diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index b2e9394..b5ea583 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -19,7 +19,7 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install package with test dependencies - run: python -m pip install --upgrade pip && pip install -e .[isapi] + run: python -m pip install --upgrade pip && pip install -e . - name: Run tests run: python -m unittest discover -s test -v diff --git a/README.md b/README.md index 5b23cf4..7d5de94 100644 --- a/README.md +++ b/README.md @@ -13,9 +13,9 @@ Code is licensed under the MIT license. * Python 3.10+ * [requests](https://pypi.org/project/requests/) >= 2.20.0 +* [xmltodict](https://pypi.org/project/xmltodict/) >= 0.13.0 Optional: -* `xmltodict` >= 0.13.0 - required for `ISAPIClient` (install with the `isapi` extra) * [pyDispatcher](https://pypi.org/project/PyDispatcher/) 2.0.5 - only needed if you want `HikCamera` to also broadcast updates via `pydispatch` signals; the built-in callback mechanism (`add_update_callback`) works without it # Installation @@ -24,12 +24,6 @@ Optional: pip install pyhik ``` -To also use `ISAPIClient`: - -``` -pip install pyhik[isapi] -``` - # Usage - Event streaming (`HikCamera`) ```python @@ -125,7 +119,34 @@ result = client.set_image_setting("Shutter/ShutterLevel", "1/500") result.success, result.sub_status # True, 'ok' ``` -A write reads the full document, changes only the target value as text and sends it back, which the devices require. Some settings conflict (enabling WDR while BLC or HLC is on, and the reverse); the setting in the way is switched off first. Firmware that reports the HLC conflict as `HLCMutexWDROrDehaze` or `WDROrDehazeMutexHLC` (HTTP 403) is handled the same way: HLC is switched off when WDR or defog is turned on, and WDR and defog are switched off when HLC is turned on. The settings that were switched off to make room are listed in `ImageWriteResult.also_changed` (for example `{"HLC/enabled": "false"}`). If a conflict cannot be resolved, `ImageWriteResult.error_description` explains it. Cameras that reject writes to the full document with `notSupport` but accept a per-feature sub-endpoint (for example `/ISAPI/Image/channels/1/Shutter`) are handled automatically when all the changes belong to the same feature (for example `BLC/enabled` and `BLC/BLCMode`). Writes are not atomic, so avoid overlapping writes to the same device. +#### Write Mechanism + +* **Read-Modify-Write:** A write operation reads the full document, changes only the target value as text, and sends the updated document back to the device. +* **Non-Atomic Operations:** Writes are **not atomic**—avoid overlapping or concurrent writes to the same device. + +#### Conflict Resolution + +Some settings inherently conflict with one another (e.g., enabling WDR while BLC or HLC is active, and vice versa). When a conflict occurs: + +* **Automatic Override:** The conflicting setting is automatically switched off first to make room for the new setting. +* **Firmware Error Codes:** Firmware reporting conflicts as `HLCMutexWDROrDehaze` or `WDROrDehazeMutexHLC` (HTTP 403) is handled identically: +* HLC is switched off when WDR or defog is turned on. +* WDR and defog are switched off when HLC is turned on. + +* **Result Tracking:** +* `ImageWriteResult.also_changed` lists the settings that were automatically disabled (e.g., `{"HLC/enabled": "false"}`). +* `ImageWriteResult.error_description` explains any conflicts that could not be resolved. + +#### Fallback Handling (Sub-Endpoints) + +* **`notSupport` Handling:** If a camera rejects writes to the full document with a `notSupport` error, the system automatically attempts to use per-feature sub-endpoints (example `/ISAPI/Image/channels/1/Shutter`). +* **Condition:** This fallback only triggers when all requested changes belong to the **same feature** (such as `BLC/enabled` and `BLC/BLCMode`). + +### Checking a device + +`python -m pyhik check` tries the image settings code against a real camera or NVR. +[TESTING.md](TESTING.md) has the instructions to use it and can be used to report a +problem. # TODO diff --git a/TESTING.md b/TESTING.md new file mode 100644 index 0000000..fa9683c --- /dev/null +++ b/TESTING.md @@ -0,0 +1,135 @@ +# Testing the image settings on a real camera or NVR + +The unit tests (`python -m unittest discover test -v`) never touch a device. This page is +for trying the code on real hardware before it is released and for reporting what +happens. `python -m pyhik check` does the work. + +## Before you start + +- Use an account that has remote configuration rights. The script never prints the + password and stops at the first authentication error. +- Everything that changes a value puts it back afterwards, even after an error. If it + could not, it says `NOT RESTORED` and tells you the value to set by hand. +- Try a camera you do not mind experimenting. The changes are small (one brightness + step, WDR and HLC toggled, or the one setting you pick with `--path`) but they are + real writes. + +```bash +git clone git@github.com:mezz64/pyHik.git +cd pyhik +python -m venv .venv && . .venv/bin/activate && pip install -e . +``` + +## Required Environment Variables + +```bash +export CAMERA_HOST=192.168.1.100 # IP address or hostname of your camera or NVR. +export CAMERA_USERNAME=admin +export CAMERA_PASSWORD='passwd123' +``` +> For security, the password is read exclusively from the environment. + +## Optional Environment Variables + +```bash +export CAMERA_PORT=80 # Custom port (default: 80; 443 enables HTTPS) +export CAMERA_SSL=1 # Enable SSL/HTTPS +export CAMERA_VERIFY_SSL=0 # Disable SSL cert verification +``` + +You may also use arguments: + +* `--host IP/hostname` (`CAMERA_HOST=...`) +* `--username admin ` (`CAMERA_USERNAME=...`) +* `--port 443` (`CAMERA_PORT=...`) +* `--ssl` (`CAMERA_SSL=1`) +* `--no-verify` (`CAMERA_VERIFY_SSL=0`) + +## What to run + +| Command | What it checks | Changes the device? | +|---|---|---| +| `python -m pyhik check` | channels, settings (with options and ranges), current values | No | +| `python -m pyhik check --write` | changes one number by one step (brightness), reads it back, restores it. With `--path` it can change a switch or a select instead (see below). If the change switched another setting off to make room, that one is put back too | Briefly | +| `python -m pyhik check --conflicts` | WDR, HLC, BLC and defog cannot all be on together. For every pair that the device offers it turns one on, then the other and prints a table of what happened. Checks that pyhik switches the other one off by itself and shows the refusal code the device used | Briefly (restores everything) | +| `python -m pyhik check --bad-password` | a wrong password is reported as an authentication failure by each client: `ISAPIAuthError` 401 from `ISAPIClient` and `HikCamera.auth_failed` | No but it makes at least two failed logins and the device may log them. Do not use it on an account close to being locked | +| `python -m pyhik check --all` | all of the above | Briefly | +| add `--proxy` / `--no-proxy` | force the NVR (ImageProxy) or the camera endpoints. Normally not needed: the script uses the NVR endpoints for an NVR or DVR, and for any device whose camera image endpoints fail while the NVR endpoints work | as above | +| add `--channel N` | use channel ID N instead of the first listed | as above | +| add `--path Some/path` | with `--write`, change this setting instead of brightness. Take the path from the list the read-only run prints (a plain `--write` also suggests a few switch and select paths). A number moves one step, a switch flips (true to false or false to true), a select takes the first option that is not the current one. Always put back afterwards | as above | +| add `-v` | print pyhik's debug log, which shows the retries and conflict handling | | + +Each check ends with `PASS`, `FAIL` or `SKIP` and the run ends with a summary. The exit +code is 0 when nothing failed, 1 when a check failed and 2 when the device could not be +reached or refused the login. + +A `SKIP` is normal: for example `--conflicts` is skipped on a device that offers fewer than +two of WDR, HLC, BLC and defog, and `--write --path` is skipped when the path is not in the +list or the setting has no other value to try. + +### Changing a switch or a select + +```bash +python -m pyhik check --channel 1 --write --path enableImageLossDetection +``` + +The script prints what it changes (`path (switch): true -> false`), the result, the value +read back, and `restore ...: restored`. Use the exact path from the read-only run: an NVR +can list settings with other names than a camera does. + +### What the `--conflicts` table looks like + +``` + already on | turn on: WDR=open WDR=auto HLC BLC=CENTER + ------------------------------------------------------------------------- + WDR=open | - - > > + HLC | > > - = + BLC=CENTER | > > = - + + = both can be on > the other one was switched off, the change worked + X refused ? could not be tried - same setting + WDR on, then HLC: WDR was switched off (by pyhik: WDR/mode) after WDROrDehazeMutexHLC + HLC on, then WDR=open: HLC was switched off (by pyhik: HLC/enabled) after HLCMutexWDROrDehaze + ... +``` + +Read it as "with the row on, turn the column on". `>` is the normal result on a camera that +has the conflict: the device refused with the code shown, pyhik switched the other setting off +and the change went through. `=` means this device allows both. `X` is a failure: the device +refused and pyhik did not know how to resolve it. That is the case to report, with the code +shown. The cases depend on which of WDR, HLC, BLC and defog your device offers; with fewer +than two the check is skipped. The camera's original values are put back at the end +(`restore: restored`). + +## NVRs + +Nothing special is needed: the script asks the device for its type and, for an NVR or DVR (or a device +whose camera endpoints fail while the NVR endpoints work), lists the attached cameras as channels and reads and writes them through the ImageProxy +endpoints. It prints `proxy endpoints: yes ...` with the reason, so you can see what it chose. +`--write` and `--conflicts` act on the first channel, or on the one given with `--channel`, +so you can test one camera behind the NVR at a time. Run the read-only check first. +If you report an issue, include if the NVR is a Hikvision or a re-branded one. + +### When the device information cannot be read + +If the device information request fails, the script prints, for example, +`device: model=? (ISAPIError: Request failed with status 503), ...` and carries on: it asks +only once, and falls back to trying the camera endpoints and then the NVR endpoints. That line +is not a failure of the check. Please include it in a report, because it has the error. + +If the error says `xmltodict is required`, the `xmltodict` package is missing from your +Python environment: run `pip install xmltodict` and try again. + +## Reporting a problem + +Open an issue (or paste into the pull request) with: + +1. The full output of the run, with `-v` for anything that failed. The script already + prints the pyhik version and the model, type and firmware of the device. Remove the + host name, IP addresses and anything else that identifies you. +2. Whether it is a camera or an NVR and whether it is rebranded. +3. What you expected and what happened, including whether the setting really changed + on the device (its own web page shows the truth). +4. The exact command you ran. + +Do not include the password, serial number or MAC address. diff --git a/pyhik/__main__.py b/pyhik/__main__.py new file mode 100644 index 0000000..93b2ffa --- /dev/null +++ b/pyhik/__main__.py @@ -0,0 +1,36 @@ +"""``python -m pyhik ``: command line tools that come with pyhik. + +Commands are imported only when they are asked for, so nothing here costs anything +for programs that use pyhik as a library. +""" + +import importlib +import sys +from typing import Optional, Sequence + +COMMANDS = { + "check": "try the image settings against a real camera or NVR", +} + + +def usage() -> str: + width = max(len(name) for name in COMMANDS) + lines = ["usage: python -m pyhik [options]", "", "commands:"] + lines += [" %-*s %s" % (width, name, text) for name, text in COMMANDS.items()] + lines += ["", "Run python -m pyhik --help for the options of a command."] + return "\n".join(lines) + + +def main(argv: Optional[Sequence[str]] = None) -> int: + args = list(sys.argv[1:] if argv is None else argv) + if not args or args[0] in ("-h", "--help"): + print(usage()) + return 0 if args else 2 + if args[0] not in COMMANDS: + print("pyhik: unknown command %r\n\n%s" % (args[0], usage()), file=sys.stderr) + return 2 + return importlib.import_module("pyhik." + args[0]).main(args[1:]) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyhik/check.py b/pyhik/check.py new file mode 100644 index 0000000..1afebc7 --- /dev/null +++ b/pyhik/check.py @@ -0,0 +1,616 @@ +"""Check pyhik's image settings support against a real camera or NVR. + +Not a unit test: unit tests must run without hardware. + +Run it with the pyhik you want to check installed: + + export CAMERA_HOST=192.168.1.100 + export CAMERA_USERNAME=admin + export CAMERA_PASSWORD='passwd123' + python -m pyhik check # read-only, always safe + +The host, port, user name, HTTPS and certificate check can also be given as --host, --port, +--username, --ssl and --no-verify, which win over the environment variables. The password is +only read from the environment. + +What to run: + + python -m pyhik check # read-only, lists every setting and its path + python -m pyhik check --write # change Color/brightnessLevel by one step and put it back + python -m pyhik check --write --path WDR/mode # a switch is flipped, a select gets another option + python -m pyhik check --conflicts # WDR / HLC / BLC / defog conflict matrix + python -m pyhik check --bad-password # a wrong password, tried by each client + python -m pyhik check --all # all of the above + python -m pyhik check -v # also show pyhik's debug log (the retries) + +--write changes the setting Color/brightnessLevel by one step (or the first numeric +setting if the device has no brightness) and restores it. To choose the setting yourself +take a path from the list printed by the read-only run. --path is only used together +with --write and works for every kind of setting: + + python -m pyhik check --write --path Color/contrastLevel # a number: one step + python -m pyhik check --write --path IrcutFilter/IrcutFilterType # a select: another option + python -m pyhik check --write --path # a switch: on <-> off + +A setting that was switched off to make room for the one under test (WDR, BLC, HLC and +defog exclude each other) is put back too and listed. + +One channel is checked per run. Without --channel, the first listed channel is used. On an NVR, +give the channel ID of the attached camera, for example the camera on channel 3: + + python -m pyhik check --channel 3 + python -m pyhik check --channel 3 --write --path Color/contrastLevel + +An NVR is detected from the device's own type (NVR or DVR). A recorder that reports the +type of a camera (IPC), or does not answer the device information request, is detected by +its camera image endpoints failing while the proxy endpoints work. Its attached cameras +are then used through the proxy endpoints. --proxy and --no-proxy override the detection. + +Optional: CAMERA_PORT (80, or 443 with HTTPS), CAMERA_SSL=1, CAMERA_VERIFY_SSL=0. + +Everything that changes a value puts it back afterwards (even after an error), and +the script says so when it could not. It stops at the first authentication error +because repeated failures can lock the account. --bad-password tries a wrong password +through each of the two clients (ISAPIClient and HikCamera); at least two failed +logins, more if a client retries; do not use it on an account that is close to being locked. +""" + +import argparse +import logging +import os +import re +import sys + +import pyhik +from pyhik.image import ImageSettingType +from pyhik.isapi import ( + ISAPIAuthError, + ISAPIClient, + ISAPIConnectionError, + ISAPIError, + ISAPINotSupportedError, +) + + +class Report: + """Collects the outcome of every check so the script can end with a summary.""" + + def __init__(self, out=print): + self.out = out + self.results = [] # (name, "PASS" | "FAIL" | "SKIP", detail) + + def add(self, name, status, detail=""): + self.results.append((name, status, detail)) + self.out(" -> %s%s" % (status, (": " + detail) if detail else "")) + + @property + def failed(self): + return any(status == "FAIL" for _, status, _ in self.results) + + def summary(self): + self.out("") + self.out("summary") + for name, status, detail in self.results: + self.out(" %-5s %s%s" % (status, name, (" (" + detail + ")") if detail else "")) + self.out(" overall: %s" % ("FAILED" if self.failed else "ok")) + + +def _flag(name, default): + return os.environ.get(name, str(int(default))).strip().lower() in ("1", "true", "yes") + + +def settings_from_env(password=None, host=None, port=None, username=None, ssl=None, verify_ssl=None): + """The connection settings. What is given here wins over the environment. + + ssl=True means HTTPS and verify_ssl=False means do not check the certificate; None means + the CAMERA_SSL / CAMERA_VERIFY_SSL variables decide. Without a port, HTTPS uses 443 and + HTTP 80. + """ + missing = [] + if not (host or os.environ.get("CAMERA_HOST")): + missing.append("CAMERA_HOST (or --host)") + if not os.environ.get("CAMERA_PASSWORD"): + missing.append("CAMERA_PASSWORD") + if missing: + sys.exit("Set these first: " + ", ".join(missing) + + " (optional: CAMERA_USERNAME or --username, CAMERA_PORT or --port," + " CAMERA_SSL or --ssl, CAMERA_VERIFY_SSL or --no-verify)") + use_ssl = bool(ssl) or _flag("CAMERA_SSL", False) + return dict( + host=host or os.environ["CAMERA_HOST"], + port=port if port is not None else int(os.environ.get("CAMERA_PORT", "443" if use_ssl else "80")), + username=username or os.environ.get("CAMERA_USERNAME", "admin"), + password=password if password is not None else os.environ["CAMERA_PASSWORD"], + ssl=use_ssl, + verify_ssl=False if verify_ssl is False else _flag("CAMERA_VERIFY_SSL", True), + ) + + +def remember_device_info(client): + """Make client.get_device_info() ask the device once per run, whether it answers or not. + + pyhik keeps an answer but not a failure, so on a device whose device info fails each of + get_device_model(), get_device_type() and get_firmware_version() sent a request of its + own and detection one more. + """ + ask = client.get_device_info + seen = {} + + def get_device_info(): + if "error" in seen: + raise seen["error"] + if "info" not in seen: + try: + seen["info"] = ask() + except Exception as err: + seen["error"] = err + raise + return seen["info"] + + client.get_device_info = get_device_info + return client + + +def client_from_env(password=None, **overrides): + return remember_device_info(ISAPIClient(**settings_from_env(password, **overrides))) + + +def _describe(err): + return "%s: %s (status=%s, sub_status=%s)" % ( + type(err).__name__, err, err.status_code, err.sub_status, + ) + + +def pick_number(settings, path=None): + """The setting to change: the given path, else brightness, else any numeric setting.""" + by_path = {s.path: s for s in settings} + if path: + return by_path.get(path) + if "Color/brightnessLevel" in by_path: + return by_path["Color/brightnessLevel"] + return next((s for s in settings if s.type is ImageSettingType.NUMBER and s.value), None) + + +def other_value(target): + """A value different from the current one to try, or None when there is none. + + A number moves one step (down at its upper limit), a switch flips and a select + takes the first other option. + """ + original = target.value + if original is None: + return None + if target.type is ImageSettingType.NUMBER: + try: + return str(int(original) + 1 if int(original) < int(target.max_value) else int(original) - 1) + except (TypeError, ValueError): + return None + if target.type is ImageSettingType.SWITCH: + return "false" if original.lower() == "true" else "true" + return next((o for o in target.options if o != original), None) + + +def suggest_paths(settings, out): + """Name a few switches and selects the user can try with --path.""" + for kind, label in ((ImageSettingType.SWITCH, "switch"), (ImageSettingType.SELECT, "select")): + paths = [s.path for s in settings if s.type is kind and s.value is not None][:3] + if paths: + out(" to try a %s too: --write --path %s" % (label, " | ".join(paths))) + + +def _short(err): + """The exception name, with the HTTP status and the device's sub status when it has them.""" + status = getattr(err, "status_code", None) + sub_status = getattr(err, "sub_status", None) + detail = ", ".join(x for x in (str(err), "status=%s" % status if status else "", + "sub_status=%s" % sub_status if sub_status else "") if x) + return "%s: %s" % (type(err).__name__, detail) + + +def _device_line(client, out): + """Print what a bug report needs to know about the device (no serial number).""" + parts = [] + for label, getter in ( + ("model", "get_device_model"), + ("type", "get_device_type"), + ("firmware", "get_firmware_version"), + ): + try: + parts.append("%s=%s" % (label, getattr(client, getter)() or "?")) + except Exception as err: # not worth failing the run for + parts.append("%s=? (%s)" % (label, _short(err))) + out("pyhik %s from %s" % (pyhik.__version__, os.path.dirname(pyhik.__file__))) + out("device: " + ", ".join(parts)) + + +# -------------------------------------------------------------------------- +# checks +# -------------------------------------------------------------------------- + + +def check_read(client, report, channel, proxy, out): + """Channels, settings and values. Returns (channel, settings) or None.""" + out("1. channels") + channels = client.get_image_channels(proxy=proxy) + for c in channels: + out(" id=%s enabled=%s input_port=%s" % (c.id, c.enabled, c.input_port)) + if not channels: + report.add("channels", "FAIL", "none listed") + return None + report.add("channels", "PASS", "%d listed" % len(channels)) + channel = channel if channel is not None else channels[0].id + + out("2. settings of channel %s" % channel) + settings = client.get_image_settings(channel, proxy=proxy) + kinds = {} + for s in settings: + kinds[s.type.value] = kinds.get(s.type.value, 0) + 1 + out(" %d settings: %s" % (len(settings), kinds)) + for s in settings: + extra = ("options=%s" % s.options) if s.options else ( + "range=%s-%s" % (s.min_value, s.max_value) if s.min_value is not None else "") + link = " (linked to %s)" % s.linked_enabled_path if s.linked_enabled_path else "" + out(" %-8s %-60s %-10s %s%s" % (s.type.value, s.path, s.value, extra, link)) + report.add("settings", "PASS" if settings else "FAIL", "%d found" % len(settings)) + + out("3. current values") + values = client.get_image_values(channel, proxy=proxy) + out(" %d values read" % len(values)) + report.add("values", "PASS" if values else "FAIL", "%d read" % len(values)) + return channel, settings + + +def check_write(client, report, channel, proxy, settings, path, out): + """Change one setting (number, switch or select), read it back, put it back.""" + out("4. change a setting and put it back") + target = pick_number(settings, path) + if target is None and path: + report.add("write", "SKIP", "no setting with the path %s (take one from the list)" % path) + return + if target is None or target.value is None: + report.add("write", "SKIP", "no suitable numeric setting (use --path)") + return + new = other_value(target) + if new is None: + report.add("write", "SKIP", "%s has no other value to try" % target.path) + return + original = target.value + before = client.get_image_values(channel, proxy=proxy) + out(" %s (%s): %s -> %s" % (target.path, target.type.value, original, new)) + try: + result = client.set_image_setting(target.path, new, channel, proxy=proxy) + out(" result: %s" % result) + if getattr(result, "also_changed", None): + out(" also switched off to make room: %s" % result.also_changed) + got = client.get_image_values(channel, proxy=proxy).get(target.path) + ok = result.success and got == new + report.add("write", "PASS" if ok else "FAIL", + "read back OK" if ok else "read back %s" % got) + finally: + result = client.set_image_setting(target.path, original, channel, proxy=proxy) + after = client.get_image_values(channel, proxy=proxy) + restored = result.success and after.get(target.path) == original + out(" restore %s -> %s: %s" % ( + target.path, original, "restored" if restored else "NOT RESTORED, set it back by hand")) + if not restored: + report.add("restore of " + target.path, "FAIL", "set it back to %s by hand" % original) + # Settings the write switched off to make room are put back as well. + for other, value in before.items(): + if other != target.path and after.get(other) != value: + back = client.set_image_setting(other, value, channel, proxy=proxy) + now = client.get_image_values(channel, proxy=proxy).get(other) + done = back.success and now == value + out(" restore %s -> %s: %s" % ( + other, value, "restored" if done else "NOT RESTORED, set it back by hand")) + if not done: + report.add("restore of " + other, "FAIL", "set it back to %s by hand" % value) + if not path: + suggest_paths(settings, out) + + +class Feature: + """A setting that can be on or off and may exclude others (WDR, HLC, BLC, defog).""" + + def __init__(self, name, variants, off, is_on, restore_on, paths): + self.name = name + self.variants = variants # [(label, apply)] ways to turn it on; apply() -> result + self.off = off # () -> result + self.is_on = is_on # (values) -> bool + self.restore_on = restore_on # (values before) -> result, puts the old "on" state back + self.paths = paths + + +def build_features(client, channel, proxy, settings): + """The conflict-prone settings this device offers, as Feature objects.""" + by_path = {s.path.lower(): s for s in settings} + + def put(path, value): + return lambda: client.set_image_setting(path, value, channel, proxy=proxy) + + features = [] + + def mode_feature(name, path): + setting = by_path.get(path) + if setting is None or "close" not in setting.options: + return + on_options = [o for o in setting.options if o != "close"] + if not on_options: + return + features.append(Feature( + name, + [("%s=%s" % (name, o), put(setting.path, o)) for o in on_options], + put(setting.path, "close"), + lambda v, p=setting.path: v.get(p) not in (None, "close"), + lambda v, p=setting.path: put(p, v[p])(), + [setting.path], + )) + + mode_feature("WDR", "wdr/mode") + mode_feature("defog", "dehaze/dehazemode") + + hlc = by_path.get("hlc/enabled") + if hlc is not None: + features.append(Feature( + "HLC", [("HLC", put(hlc.path, "true"))], put(hlc.path, "false"), + lambda v, p=hlc.path: v.get(p) == "true", + lambda v, p=hlc.path: put(p, "true")(), [hlc.path], + )) + + blc = by_path.get("blc/blcmode") + if blc is not None and blc.linked_enabled_path and blc.off_value is not None: + modes = [o for o in blc.options if o != blc.off_value] + preferred = next((o for o in modes if o.lower() in ("auto", "center")), None) + mode = preferred or (modes[0] if modes else None) + if mode is not None: + def blc_set(value, blc=blc): + return client.set_image_linked_setting(blc, value, channel, proxy=proxy) + + features.append(Feature( + "BLC", [("BLC=%s" % mode, lambda: blc_set(mode))], lambda: blc_set(blc.off_value), + lambda v, e=blc.linked_enabled_path: v.get(e) == "true", + lambda v, b=blc: blc_set(v.get(b.path) or mode), + [blc.path, blc.linked_enabled_path], + )) + return features + + +class ConflictCodes(logging.Handler): + """Collects the refusal codes pyhik logs when it resolves a conflict.""" + + def __init__(self): + super().__init__(logging.INFO) + self.codes = [] + + def emit(self, record): + match = re.match(r"Conflict \((\w+)\)", record.getMessage()) + if match: + self.codes.append(match.group(1)) + + +def check_conflicts(client, report, channel, proxy, settings, out): + """Try every "turn A on while B is on" combination and show how the device answers.""" + out("5. conflicts between WDR, HLC, BLC and defog (each is turned on while another is on)") + features = build_features(client, channel, proxy, settings) + if len(features) < 2: + report.add("conflicts", "SKIP", "the device offers fewer than two of WDR, HLC, BLC and defog") + return + + def read(): + return client.get_image_values(channel, proxy=proxy) + + before = read() + out(" before: " + ", ".join( + "%s=%s" % (p, before.get(p)) for f in features for p in f.paths)) + codes = ConflictCodes() + logger = logging.getLogger("pyhik") + old_level = logger.level + logger.addHandler(codes) + logger.setLevel(min(old_level or logging.INFO, logging.INFO)) + + columns = [(f, label, apply) for f in features for label, apply in f.variants] + cells = {} # (row feature name, column label) -> (symbol, detail) + try: + for row in features: + for feature, label, apply in columns: + if feature is row: + cells[(row.name, label)] = ("-", "") + continue + for f in features: # known starting point: everything off + f.off() + first = row.variants[0] + setup = first[1]() + if not setup.success: + cells[(row.name, label)] = ("?", "could not turn %s on first: %s" % ( + first[0], setup.error_description)) + continue + codes.codes.clear() + result = apply() + values = read() + row_still_on = row.is_on(values) + seen = ", ".join(dict.fromkeys(codes.codes)) + if result.success and row_still_on: + cells[(row.name, label)] = ("=", "both can be on") + elif result.success: + detail = "%s was switched off" % row.name + if result.also_changed: + detail += " (by pyhik: %s)" % ", ".join(sorted(result.also_changed)) + cells[(row.name, label)] = (">", detail + (" after %s" % seen if seen else "")) + else: + cells[(row.name, label)] = ("X", "refused: %s" % ( + result.sub_status if not seen else "%s (%s)" % (result.sub_status, seen))) + finally: + logger.removeHandler(codes) + logger.setLevel(old_level) + for f in features: + f.off() + for f in features: + if f.is_on(before): + f.restore_on(before) + after = read() + restored = all(after.get(p) == before.get(p) for f in features for p in f.paths) + out(" restore: %s" % ("restored" if restored else "NOT RESTORED, put these back by hand: " + ", ".join( + "%s=%s" % (p, before.get(p)) for f in features for p in f.paths if after.get(p) != before.get(p)))) + if not restored: + report.add("restore of conflict settings", "FAIL", "see above") + + width = max(len(label) for _, label, _ in columns) + 2 + out("") + out(" %-14s| turn on: %s" % ("already on", "".join("%-*s" % (width, label) for _, label, _ in columns))) + out(" %s" % ("-" * (14 + 11 + width * len(columns)))) + for row in features: + out(" %-14s| %s" % ( + row.variants[0][0], "".join("%-*s" % (width, cells[(row.name, label)][0]) for _, label, _ in columns))) + out("") + out(" = both can be on > the other one was switched off, the change worked") + out(" X refused ? could not be tried - same setting") + for (row_name, label), (symbol, detail) in cells.items(): + if symbol in "=>X?": + out(" %s on, then %s: %s" % (row_name, label, detail)) + + refused = [(r, c) for (r, c), (symbol, _) in cells.items() if symbol == "X"] + tried = [v for v in cells.values() if v[0] != "-"] + resolved = sum(1 for v in tried if v[0] == ">") + summary = "%d cases: %d fine, %d resolved by switching the other off, %d refused" % ( + len(tried), sum(1 for v in tried if v[0] == "="), resolved, len(refused)) + for r, c in refused: + report.add("conflict: %s on, then %s" % (r, c), "FAIL", cells[(r, c)][1]) + report.add("conflicts", "FAIL" if refused else "PASS", summary) + + +def check_bad_password(report, out, overrides=None): + """A wrong password must be reported as an authentication failure.""" + out("6. wrong password, tried by each client (the device may log these attempts)") + cfg = settings_from_env(password=os.environ["CAMERA_PASSWORD"] + "-wrong", **(overrides or {})) + try: + client = ISAPIClient(**cfg) + try: + client.get_image_channels() + finally: + client.close() + except ISAPIAuthError as err: + report.add("wrong password (ISAPIClient)", + "PASS" if err.status_code == 401 else "FAIL", + "ISAPIAuthError status=%s" % err.status_code) + except ISAPIError as err: + report.add("wrong password (ISAPIClient)", "FAIL", _describe(err)) + else: + report.add("wrong password (ISAPIClient)", "FAIL", "the device accepted a wrong password") + + from pyhik.hikvision import HikCamera + + url = ("https" if cfg["ssl"] else "http") + "://" + cfg["host"] + camera = HikCamera(url, cfg["port"], cfg["username"], cfg["password"], cfg["verify_ssl"]) + if not hasattr(camera, "auth_failed"): + report.add("wrong password (HikCamera.auth_failed)", "SKIP", + "this pyhik has no auth_failed (pyHik PR 6)") + elif camera.get_id is None and camera.auth_failed: + report.add("wrong password (HikCamera.auth_failed)", "PASS") + else: + report.add("wrong password (HikCamera.auth_failed)", "FAIL", + "device id=%s auth_failed=%s" % (camera.get_id, camera.auth_failed)) + + +def detect_proxy(client, proxy, out): + """Decide whether to use the proxy endpoints (unless --proxy/--no-proxy says). + + An NVR or DVR is recognised by its device type. That is not always enough: some + recorders report the type of a camera (IPC, for example), some do not answer the + device information request at all, and real cameras answer both. So unless the + type says NVR or DVR, the camera endpoints are tried first, and if they fail with + a device error the proxy endpoints are tried before giving up. + """ + if proxy is not None: + out("proxy endpoints: %s (set on the command line)" % ("yes" if proxy else "no")) + return proxy + try: + device_type = str(client.get_device_type()).strip() + except Exception as err: # the endpoints below can still tell + device_type = None + out("device type: unknown (could not read it: %s)" % type(err).__name__) + if (device_type or "").upper() in ("NVR", "DVR"): + out("proxy endpoints: yes (device type %s)" % device_type) + return True + try: + client.get_image_channels(proxy=False) + except (ISAPIAuthError, ISAPIConnectionError): + pass # not a reason to try other endpoints; the checks report it + except ISAPIError as err: + out("camera image endpoints failed (%s); trying the proxy endpoints" % _describe(err)) + try: + if client.get_image_channels(proxy=True): + out("proxy endpoints: yes (the proxy endpoints work)") + return True + except ISAPIError: + pass + out("proxy endpoints: no (device type %s)" % (device_type or "unknown")) + return False + + +def run(client, channel=None, proxy=None, write=False, path=None, conflicts=False, + bad_password=False, out=print, report=None, overrides=None): + """Return 0 if every check passed, 1 otherwise. + + overrides: host, port and username given on the command line, for --bad-password. + """ + report = report or Report(out) + _device_line(client, out) + proxy = detect_proxy(client, proxy, out) + found = check_read(client, report, channel, proxy, out) + if found is not None: + channel, settings = found + if write: + check_write(client, report, channel, proxy, settings, path, out) + if conflicts: + check_conflicts(client, report, channel, proxy, settings, out) + if bad_password: + check_bad_password(report, out, overrides) + if not (write or conflicts or bad_password): + out("read-only checks done; see --help for the checks that change a value") + report.summary() + return 1 if report.failed else 0 + + +def main(argv=None): + parser = argparse.ArgumentParser(prog="python -m pyhik check", description=__doc__.split("\n")[0]) + parser.add_argument("--host", help="host name or IP address (default: CAMERA_HOST)") + parser.add_argument("--port", type=int, help="port (default: CAMERA_PORT, else 80)") + parser.add_argument("--username", help="user name (default: CAMERA_USERNAME, else admin)") + parser.add_argument("--ssl", action="store_true", + help="use HTTPS, on port 443 unless --port says otherwise (default: CAMERA_SSL)") + parser.add_argument("--no-verify", action="store_true", + help="do not check the device's certificate, for a self-signed one (default: CAMERA_VERIFY_SSL=0)") + parser.add_argument("--proxy", dest="proxy", action="store_true", default=None, + help="force the NVR endpoints (normally detected)") + parser.add_argument("--no-proxy", dest="proxy", action="store_false", + help="force the camera endpoints (normally detected)") + parser.add_argument("--channel", type=int, help="channel ID (default: first listed)") + parser.add_argument("--write", action="store_true", help="change one setting and restore it (a number by one step; with --path also a switch or a select)") + parser.add_argument("--path", help="setting to change with --write: a number, switch or select (default Color/brightnessLevel)") + parser.add_argument("--conflicts", action="store_true", + help="try every WDR / HLC / BLC / defog combination and print a table (restores them)") + parser.add_argument("--bad-password", action="store_true", + help="try a wrong password through each client (at least two failed logins)") + parser.add_argument("--all", action="store_true", help="--write --conflicts --bad-password") + parser.add_argument("-v", "--verbose", action="store_true", help="show pyhik's debug log") + args = parser.parse_args(argv) + if args.verbose: + logging.basicConfig(level=logging.DEBUG, format=" [%(name)s] %(message)s") + logging.getLogger("urllib3").setLevel(logging.WARNING) + overrides = dict(host=args.host, port=args.port, username=args.username, + ssl=True if args.ssl else None, verify_ssl=False if args.no_verify else None) + client = client_from_env(**overrides) + try: + return run(client, args.channel, args.proxy, args.write or args.all, args.path, + args.conflicts or args.all, args.bad_password or args.all, + overrides=overrides) + except ISAPIAuthError as err: + print("authentication/permission problem, stopping: " + _describe(err)) + except ISAPINotSupportedError as err: + print("not supported by this device: " + _describe(err)) + except (ISAPIConnectionError, ISAPIError) as err: + print("request failed: " + _describe(err)) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyhik/isapi.py b/pyhik/isapi.py index 9dc9de6..9ad7a80 100644 --- a/pyhik/isapi.py +++ b/pyhik/isapi.py @@ -16,6 +16,7 @@ from urllib.parse import quote, urlparse, urlunparse import requests +import xmltodict from requests.auth import HTTPBasicAuth, HTTPDigestAuth from requests.structures import CaseInsensitiveDict @@ -30,11 +31,6 @@ parse_response_status, ) -try: - import xmltodict -except ImportError: - xmltodict = None - _LOGGER = logging.getLogger(__name__) # ISAPI Endpoints @@ -347,11 +343,6 @@ def _check_digest_auth(self) -> Optional[HTTPDigestAuth]: def _parse_xml(self, text: str) -> Dict[str, Any]: """Parse XML response to dictionary.""" - if xmltodict is None: - raise ISAPIError( - "xmltodict is required for ISAPI client. " - "Install it with: pip install xmltodict" - ) try: return xmltodict.parse(text) except Exception: @@ -359,11 +350,6 @@ def _parse_xml(self, text: str) -> Dict[str, Any]: def _unparse_xml(self, data: Dict[str, Any]) -> str: """Convert dictionary to XML string.""" - if xmltodict is None: - raise ISAPIError( - "xmltodict is required for ISAPI client. " - "Install it with: pip install xmltodict" - ) return xmltodict.unparse(data) def request( diff --git a/setup.py b/setup.py index 3261bb4..6193997 100755 --- a/setup.py +++ b/setup.py @@ -41,8 +41,11 @@ python_requires='>=3.10', install_requires=[ 'requests>=2.20.0', + 'xmltodict>=0.13.0', ], extras_require={ - 'isapi': ['xmltodict>=0.13.0'], + # xmltodict is now always installed; the extra is kept so that + # "pip install pyhik[isapi]" keeps working. + 'isapi': [], }, ) diff --git a/test/test_check.py b/test/test_check.py new file mode 100644 index 0000000..c94de51 --- /dev/null +++ b/test/test_check.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +"""Tests for pyhik.check, the live check run by hand against a real device.""" + +import unittest +from unittest.mock import patch + +from pyhik import __main__ as pyhik_main +from pyhik import check +from pyhik.image import ImageSetting, ImageSettingType as T +from pyhik.isapi import ISAPIAuthError, ISAPIConnectionError, ISAPIError + +E503 = ISAPIError("Request failed with status 503", 503, "serviceUnavailable") +E400 = ISAPIError("bad", 400, "badXmlContent") + + +class FakeDetect: + """Answers the two calls detect_proxy makes.""" + + def __init__(self, device_type, camera, proxy): + self.device_type, self.camera, self.proxy = device_type, camera, proxy + self.calls = [] + + def get_device_type(self): + if isinstance(self.device_type, Exception): + raise self.device_type + return self.device_type + + def get_image_channels(self, proxy=False): + self.calls.append(proxy) + answer = self.proxy if proxy else self.camera + if isinstance(answer, Exception): + raise answer + return answer + + +class TestDetectProxy(unittest.TestCase): + def detect(self, client, flag=None): + return check.detect_proxy(client, flag, lambda line: None) + + def test_decisions(self): + cases = ( + ("type unreadable, camera 503, proxy lists", FakeDetect(E503, E503, [1]), None, True), + ("type unreadable, camera 400, proxy lists", FakeDetect(E503, E400, [1]), None, True), + ("type unreadable, camera works", FakeDetect(E503, [1], E503), None, False), + ("type unreadable, camera 401", FakeDetect(E503, ISAPIAuthError("x", 401), [1]), None, False), + ("type unreadable, connection error", FakeDetect(E503, ISAPIConnectionError("x"), [1]), None, False), + ("type unreadable, both fail", FakeDetect(E503, E503, E503), None, False), + ("type IPC, camera 503, proxy lists", FakeDetect("IPC", E503, [1]), None, True), + ("type IPC, camera works", FakeDetect("IPC", [1], [1]), None, False), + ("type NVR", FakeDetect("NVR", E503, [1]), None, True), + ("type DVR", FakeDetect("dvr", E503, [1]), None, True), + ("--no-proxy wins", FakeDetect("NVR", E503, [1]), False, False), + ("--proxy wins", FakeDetect("IPC", [1], E503), True, True), + ) + for name, client, flag, expected in cases: + with self.subTest(name): + self.assertIs(self.detect(client, flag), expected) + + def test_an_nvr_type_needs_no_probe(self): + client = FakeDetect("NVR", E503, [1]) + self.detect(client) + self.assertEqual(client.calls, []) + + +class Result: + def __init__(self, success=True, also_changed=None): + self.success = success + self.also_changed = also_changed or {} + + def __str__(self): + return "ok" if self.success else "failed" + + +class FakeDevice: + """Keeps values; turning WDR on switches HLC off, as a device that has the conflict.""" + + def __init__(self, values, fail_restore=False): + self.values = dict(values) + self.sets = [] + self.fail_restore = fail_restore + + def get_image_values(self, channel=None, proxy=False): + return dict(self.values) + + def set_image_setting(self, path, value, channel=None, proxy=False): + self.sets.append((path, value)) + if self.fail_restore and len(self.sets) > 1 and path == "Color/brightnessLevel": + return Result(False) + also = {} + if path == "WDR/enabled" and value == "true" and self.values.get("HLC/enabled") == "true": + self.values["HLC/enabled"] = "false" + also = {"HLC/enabled": "false"} + self.values[path] = value + return Result(True, also) + + +SETTINGS = [ + ImageSetting("Color/brightnessLevel", T.NUMBER, min_value=0, max_value=100, value="50"), + ImageSetting("Color/contrastLevel", T.NUMBER, min_value=0, max_value=100, value="100"), + ImageSetting("WDR/enabled", T.SWITCH, value="false"), + ImageSetting("HLC/enabled", T.SWITCH, value="true"), + ImageSetting("IrcutFilter/IrcutFilterType", T.SELECT, options=["auto", "day", "night"], value="auto"), + ImageSetting("OnlyOne/mode", T.SELECT, options=["a"], value="a"), + ImageSetting("Unset/x", T.NUMBER, min_value=0, max_value=5, value=None), +] +VALUES = {s.path: s.value for s in SETTINGS if s.value is not None} + + +class TestCheckWrite(unittest.TestCase): + def write(self, path, device=None): + device = device or FakeDevice(VALUES) + lines = [] + report = check.Report(lines.append) + check.check_write(device, report, 1, False, SETTINGS, path, lines.append) + return device, report, lines + + def test_default_is_a_number_moved_one_step_and_restored(self): + device, report, lines = self.write(None) + self.assertEqual(device.sets, [("Color/brightnessLevel", "51"), ("Color/brightnessLevel", "50")]) + self.assertEqual(device.values, VALUES) + self.assertFalse(report.failed) + self.assertTrue(any("--path" in line for line in lines), "names paths to try") + + def test_number_at_its_maximum_goes_down(self): + device, _, _ = self.write("Color/contrastLevel") + self.assertEqual(device.sets[0], ("Color/contrastLevel", "99")) + self.assertEqual(device.values, VALUES) + + def test_select_takes_another_option(self): + device, report, _ = self.write("IrcutFilter/IrcutFilterType") + self.assertEqual(device.sets[0], ("IrcutFilter/IrcutFilterType", "day")) + self.assertEqual(device.values, VALUES) + self.assertFalse(report.failed) + + def test_switch_is_flipped(self): + device, _, _ = self.write("HLC/enabled") + self.assertEqual(device.sets[0], ("HLC/enabled", "false")) + self.assertEqual(device.values, VALUES) + + def test_setting_switched_off_to_make_room_is_put_back(self): + device, report, lines = self.write("WDR/enabled") + self.assertEqual(device.values, VALUES) + self.assertIn(("HLC/enabled", "true"), device.sets) + self.assertTrue(any("also switched off" in line for line in lines)) + self.assertFalse(report.failed) + + def test_nothing_to_try_is_skipped(self): + for path in ("OnlyOne/mode", "No/such", "Unset/x"): + with self.subTest(path): + device, report, _ = self.write(path) + self.assertEqual(device.sets, []) + self.assertFalse(report.failed) + + def test_failed_restore_is_reported(self): + _, report, lines = self.write(None, FakeDevice(VALUES, fail_restore=True)) + self.assertTrue(report.failed) + self.assertTrue(any("NOT RESTORED" in line for line in lines)) + + +class TestOtherValue(unittest.TestCase): + def test_each_kind(self): + by_path = {s.path: s for s in SETTINGS} + self.assertEqual(check.other_value(by_path["Color/brightnessLevel"]), "51") + self.assertEqual(check.other_value(by_path["Color/contrastLevel"]), "99") + self.assertEqual(check.other_value(by_path["WDR/enabled"]), "true") + self.assertEqual(check.other_value(by_path["HLC/enabled"]), "false") + self.assertEqual(check.other_value(by_path["IrcutFilter/IrcutFilterType"]), "day") + self.assertIsNone(check.other_value(by_path["OnlyOne/mode"])) + + +ENV = {"CAMERA_HOST": "env-host", "CAMERA_PORT": "8080", "CAMERA_USERNAME": "env-user", "CAMERA_PASSWORD": "pw"} + + +class TestSettings(unittest.TestCase): + def test_environment_only(self): + with patch.dict("os.environ", ENV, clear=True): + cfg = check.settings_from_env() + self.assertEqual((cfg["host"], cfg["port"], cfg["username"], cfg["password"]), + ("env-host", 8080, "env-user", "pw")) + + def test_defaults(self): + with patch.dict("os.environ", {"CAMERA_HOST": "h", "CAMERA_PASSWORD": "pw"}, clear=True): + cfg = check.settings_from_env() + self.assertEqual((cfg["port"], cfg["username"], cfg["ssl"], cfg["verify_ssl"]), (80, "admin", False, True)) + + def test_command_line_wins_over_the_environment(self): + with patch.dict("os.environ", ENV, clear=True): + cfg = check.settings_from_env(host="arg-host", port=81, username="arg-user") + self.assertEqual((cfg["host"], cfg["port"], cfg["username"]), ("arg-host", 81, "arg-user")) + + def test_https_uses_port_443_unless_a_port_is_given(self): + base = {"CAMERA_HOST": "h", "CAMERA_PASSWORD": "pw"} + with patch.dict("os.environ", base, clear=True): + self.assertEqual(check.settings_from_env(ssl=True)["port"], 443) + self.assertEqual(check.settings_from_env(ssl=True, port=8443)["port"], 8443) + self.assertEqual(check.settings_from_env()["port"], 80) + with patch.dict("os.environ", dict(base, CAMERA_SSL="1"), clear=True): + cfg = check.settings_from_env() + self.assertEqual((cfg["ssl"], cfg["port"]), (True, 443)) + + def test_ssl_and_certificate_options_win_over_the_environment(self): + with patch.dict("os.environ", dict(ENV, CAMERA_SSL="0", CAMERA_VERIFY_SSL="1"), clear=True): + cfg = check.settings_from_env(ssl=True, verify_ssl=False) + self.assertEqual((cfg["ssl"], cfg["verify_ssl"]), (True, False)) + cfg = check.settings_from_env() + self.assertEqual((cfg["ssl"], cfg["verify_ssl"]), (False, True)) + with patch.dict("os.environ", dict(ENV, CAMERA_VERIFY_SSL="0"), clear=True): + self.assertFalse(check.settings_from_env()["verify_ssl"]) + + def test_host_on_the_command_line_is_enough(self): + with patch.dict("os.environ", {"CAMERA_PASSWORD": "pw"}, clear=True): + self.assertEqual(check.settings_from_env(host="h")["host"], "h") + + def test_password_only_comes_from_the_environment(self): + with patch.dict("os.environ", ENV, clear=True): + self.assertEqual(check.settings_from_env(password="other")["password"], "other") + with patch.dict("os.environ", {"CAMERA_HOST": "h"}, clear=True): + with self.assertRaises(SystemExit) as raised: + check.settings_from_env() + self.assertIn("CAMERA_PASSWORD", str(raised.exception)) + + +class TestMain(unittest.TestCase): + def test_asks_for_the_environment_variables(self): + with patch.dict("os.environ", {}, clear=True): + with self.assertRaises(SystemExit) as raised: + check.main([]) + self.assertIn("CAMERA_HOST", str(raised.exception)) + self.assertIn("--host", str(raised.exception)) + + def test_host_port_and_username_reach_the_client_and_the_checks(self): + with patch.object(check, "client_from_env", return_value="client") as client, \ + patch.object(check, "run", return_value=0) as run: + self.assertEqual(check.main(["--host", "h", "--port", "81", "--username", "u", "--bad-password"]), 0) + expected = {"host": "h", "port": 81, "username": "u", "ssl": None, "verify_ssl": None} + client.assert_called_once_with(**expected) + self.assertEqual(run.call_args.kwargs["overrides"], expected) + + def test_ssl_and_no_verify_reach_the_client_and_the_checks(self): + with patch.object(check, "client_from_env", return_value="client") as client, \ + patch.object(check, "run", return_value=0) as run: + check.main(["--ssl", "--no-verify"]) + expected = {"host": None, "port": None, "username": None, "ssl": True, "verify_ssl": False} + client.assert_called_once_with(**expected) + self.assertEqual(run.call_args.kwargs["overrides"], expected) + + def test_not_given_means_the_environment(self): + with patch.object(check, "client_from_env", return_value="client") as client, \ + patch.object(check, "run", return_value=0): + check.main([]) + client.assert_called_once_with(host=None, port=None, username=None, ssl=None, verify_ssl=None) + +class TestPackageCommand(unittest.TestCase): + """python -m pyhik """ + + def run_main(self, argv): + with patch("builtins.print"): + return pyhik_main.main(argv) + + def test_no_command_prints_the_usage(self): + self.assertEqual(self.run_main([]), 2) + self.assertEqual(self.run_main(["--help"]), 0) + + def test_unknown_command(self): + self.assertEqual(self.run_main(["nope"]), 2) + + def test_check_gets_the_rest_of_the_arguments(self): + with patch.object(check, "main", return_value=7) as run: + self.assertEqual(pyhik_main.main(["check", "--write", "--path", "WDR/enabled"]), 7) + run.assert_called_once_with(["--write", "--path", "WDR/enabled"]) + + def test_check_help(self): + with patch("sys.stdout"), self.assertRaises(SystemExit) as raised: + pyhik_main.main(["check", "--help"]) + self.assertEqual(raised.exception.code, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_isapi.py b/test/test_isapi.py index cdbf075..42f1f9a 100644 --- a/test/test_isapi.py +++ b/test/test_isapi.py @@ -7,9 +7,6 @@ import requests from requests.auth import HTTPBasicAuth, HTTPDigestAuth -# xmltodict is an optional extra (pip install pyHik[isapi]); tests that need -# response parsing are skipped when it isn't installed. -from pyhik.isapi import xmltodict from pyhik.isapi import ( ISAPIClient, ISAPIError, @@ -181,7 +178,6 @@ def test_context_manager(self): class TestISAPIClientRequests(unittest.TestCase): """Test ISAPIClient HTTP request handling.""" - @unittest.skipIf(xmltodict is None, "requires the isapi extra") def test_auth_detection_digest(self, mock_session_class): """Test digest auth detection.""" session = mock_session_class.return_value