From 27e6542eee8bb4bd338e1ad365818ab08b57537a Mon Sep 17 00:00:00 2001 From: Charles Graham Date: Tue, 14 Apr 2026 12:41:30 -0500 Subject: [PATCH 1/7] Fix blob download unsafe default destination --- cwmscli/commands/blob.py | 12 +++++- tests/commands/test_blob_upload.py | 60 ++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/cwmscli/commands/blob.py b/cwmscli/commands/blob.py index 16fb31a..4e4d3f2 100644 --- a/cwmscli/commands/blob.py +++ b/cwmscli/commands/blob.py @@ -103,6 +103,16 @@ def _save_blob_content( return dest +def _default_download_dest(blob_id: str) -> str: + target = blob_id.lstrip("/\\") + if not target: + raise ValueError( + f"Blob ID must include a non-root destination name. " + f"Pass --dest explicitly if needed. Docs: {BLOB_DOCS_URL}" + ) + return target + + def _blob_media_type(cwms_module, office: str, blob_id: str) -> Optional[str]: try: result = cwms_module.get_blobs(office_id=office, blob_id_like=blob_id) @@ -577,7 +587,7 @@ def download_cmd( try: blob_content = cwms.get_blob(office_id=office, blob_id=bid) - target = dest or bid + target = dest or _default_download_dest(bid) _save_blob_content( blob_content, dest=target, diff --git a/tests/commands/test_blob_upload.py b/tests/commands/test_blob_upload.py index 09a370c..84fae93 100644 --- a/tests/commands/test_blob_upload.py +++ b/tests/commands/test_blob_upload.py @@ -6,6 +6,7 @@ from cwmscli.commands.blob import ( _blob_id_for_path, + _default_download_dest, _find_blob_id_collisions, _list_matching_files, _save_blob_content, @@ -160,6 +161,11 @@ def test_save_blob_content_writes_raw_text(tmp_path): assert dest.read_text(encoding="utf-8") == "plain text payload" +def test_default_download_dest_strips_leading_path_separators(): + assert _default_download_dest("/REPORTS/REL-BLB") == "REPORTS/REL-BLB" + assert _default_download_dest("\\REPORTS\\REL-BLB") == "REPORTS\\REL-BLB" + + def test_download_cmd_uses_media_type_to_write_text(tmp_path, monkeypatch): dest = tmp_path / "downloaded" @@ -208,6 +214,60 @@ class FakeHTTPError(Exception): assert saved.read_text(encoding="utf-8") == "retrieved text" +def test_download_cmd_default_dest_stays_relative_for_leading_slash_id( + tmp_path, monkeypatch +): + class FakeBlobListing: + df = pd.DataFrame( + [ + { + "id": "/REPORTS/REL-BLB", + "media-type-id": "text/plain", + "description": "x", + } + ] + ) + + class FakeCwms: + @staticmethod + def init_session(api_root, api_key=None): + return None + + @staticmethod + def get_blob(office_id, blob_id): + assert blob_id == "/REPORTS/REL-BLB" + return "retrieved text" + + @staticmethod + def get_blobs(office_id, blob_id_like): + assert blob_id_like == "/REPORTS/REL-BLB" + return FakeBlobListing() + + monkeypatch.setitem(sys.modules, "cwms", FakeCwms) + + class FakeHTTPError(Exception): + pass + + monkeypatch.setitem( + sys.modules, "requests", types.SimpleNamespace(HTTPError=FakeHTTPError) + ) + + monkeypatch.chdir(tmp_path) + + download_cmd( + blob_id="/reports/rel-blb", + dest=None, + office="SWT", + api_root="https://example.test/", + api_key="x", + dry_run=False, + ) + + saved = tmp_path / "REPORTS" / "REL-BLB.txt" + assert saved.exists() + assert saved.read_text(encoding="utf-8") == "retrieved text" + + def test_download_cmd_initializes_session_with_api_key(tmp_path, monkeypatch): dest = tmp_path / "downloaded" calls = [] From c3811245a8a8678153487a5e3502dd5ed8c4b65c Mon Sep 17 00:00:00 2001 From: Charles Graham Date: Tue, 14 Apr 2026 12:46:29 -0500 Subject: [PATCH 2/7] Fix clob download unsafe default destination --- cwmscli/commands/clob.py | 12 ++++++++- tests/commands/test_clob.py | 53 ++++++++++++++++++++++++++++++++++++- 2 files changed, 63 insertions(+), 2 deletions(-) diff --git a/cwmscli/commands/clob.py b/cwmscli/commands/clob.py index a2584b9..877aca8 100644 --- a/cwmscli/commands/clob.py +++ b/cwmscli/commands/clob.py @@ -29,6 +29,16 @@ def _write_clob_content(content: str, dest: str) -> str: return dest +def _default_download_dest(clob_id: str) -> str: + target = clob_id.lstrip("/\\") + if not target: + raise ValueError( + "Clob ID must include a non-root destination name. " + "Pass --dest explicitly if needed." + ) + return target + + def _clob_endpoint_id(clob_id: str) -> tuple[str, Optional[str]]: normalized = clob_id.upper() if has_invalid_chars(normalized): @@ -198,7 +208,7 @@ def download_cmd( content = str(payload) else: content = _get_special_clob_text(office=office, clob_id=query_id) - target = dest or bid + target = dest or _default_download_dest(bid) _write_clob_content(content, target) logging.info(f"Downloaded clob to: {target}") except requests.HTTPError as e: diff --git a/tests/commands/test_clob.py b/tests/commands/test_clob.py index d32ef8d..727c143 100644 --- a/tests/commands/test_clob.py +++ b/tests/commands/test_clob.py @@ -6,6 +6,7 @@ from cwmscli.commands import commands_cwms from cwmscli.commands.clob import ( _clob_endpoint_id, + _default_download_dest, delete_cmd, download_cmd, list_cmd, @@ -36,6 +37,11 @@ def test_clob_endpoint_id_uses_ignored_path_for_special_chars(): assert _clob_endpoint_id("path/id") == ("ignored", "PATH/ID") +def test_default_download_dest_strips_leading_path_separators(): + assert _default_download_dest("/REPORTS/REL-CLOB") == "REPORTS/REL-CLOB" + assert _default_download_dest("\\REPORTS\\REL-CLOB") == "REPORTS\\REL-CLOB" + + def test_download_cmd_uses_default_dest_and_writes_text(tmp_path, monkeypatch): calls = [] @@ -87,6 +93,52 @@ class FakeHTTPError(Exception): ] +def test_download_cmd_default_dest_stays_relative_for_leading_slash_id( + tmp_path, monkeypatch +): + class FakeClobResponse: + json = {"value": "retrieved clob text"} + + class FakeCwms: + @staticmethod + def init_session(api_root, api_key): + return None + + @staticmethod + def get_clob(office_id, clob_id): + assert clob_id == "/REPORTS/REL-CLOB" + return FakeClobResponse() + + monkeypatch.setitem(sys.modules, "cwms", FakeCwms) + monkeypatch.setattr("cwmscli.commands.clob.cwms", FakeCwms) + + class FakeHTTPError(Exception): + pass + + monkeypatch.setitem( + sys.modules, "requests", types.SimpleNamespace(HTTPError=FakeHTTPError) + ) + monkeypatch.setattr( + "cwmscli.commands.clob.requests", + types.SimpleNamespace(HTTPError=FakeHTTPError), + ) + + monkeypatch.chdir(tmp_path) + + download_cmd( + clob_id="/reports/rel-clob", + dest=None, + office="SWT", + api_root="https://example.test/", + api_key="apikey 123", + dry_run=False, + ) + + saved = tmp_path / "REPORTS" / "REL-CLOB" + assert saved.exists() + assert saved.read_text(encoding="utf-8") == "retrieved clob text" + + def test_download_cmd_uses_query_override_for_special_char_ids(tmp_path, monkeypatch): calls = [] @@ -196,7 +248,6 @@ class FakeHTTPError(Exception): assert calls == [("init_session", "https://example.test/", None)] - def test_list_cmd_initializes_session_with_api_key(monkeypatch): calls = [] From 8e778dd3e67f8f4f9a468a9985358295935390cf Mon Sep 17 00:00:00 2001 From: Charles Graham Date: Tue, 14 Apr 2026 12:48:37 -0500 Subject: [PATCH 3/7] Fix local download error handling for blob and clob --- cwmscli/commands/blob.py | 16 ++++----- cwmscli/commands/clob.py | 16 ++++----- cwmscli/utils/__init__.py | 12 +++++++ tests/commands/test_blob_upload.py | 52 +++++++++++++++++++++++++++++ tests/commands/test_clob.py | 53 ++++++++++++++++++++++++++++++ 5 files changed, 131 insertions(+), 18 deletions(-) diff --git a/cwmscli/commands/blob.py b/cwmscli/commands/blob.py index 4e4d3f2..781df0a 100644 --- a/cwmscli/commands/blob.py +++ b/cwmscli/commands/blob.py @@ -9,7 +9,12 @@ from collections import defaultdict from typing import Optional, Sequence, Tuple, Union -from cwmscli.utils import colors, get_api_key, log_scoped_read_hint +from cwmscli.utils import ( + colors, + format_local_download_error, + get_api_key, + log_scoped_read_hint, +) from cwmscli.utils.click_help import DOCS_BASE_URL from cwmscli.utils.deps import requires @@ -606,14 +611,7 @@ def download_cmd( ) sys.exit(1) except Exception as e: - logging.error(f"Failed to download: {e}") - log_scoped_read_hint( - api_key=resolved_api_key, - anonymous=anonymous, - office=office, - action="download", - resource="blob content", - ) + logging.error(format_local_download_error(e, BLOB_DOCS_URL)) sys.exit(1) diff --git a/cwmscli/commands/clob.py b/cwmscli/commands/clob.py index 877aca8..410bf0f 100644 --- a/cwmscli/commands/clob.py +++ b/cwmscli/commands/clob.py @@ -9,7 +9,12 @@ import requests from cwms import api as cwms_api -from cwmscli.utils import get_api_key, has_invalid_chars, log_scoped_read_hint +from cwmscli.utils import ( + format_local_download_error, + get_api_key, + has_invalid_chars, + log_scoped_read_hint, +) def _join_api_url(api_root: str, path: str) -> str: @@ -223,14 +228,7 @@ def download_cmd( ) sys.exit(1) except Exception as e: - logging.error(f"Failed to download: {e}") - log_scoped_read_hint( - api_key=resolved_api_key, - anonymous=anonymous, - office=office, - action="download", - resource="clob content", - ) + logging.error(format_local_download_error(e, "")) sys.exit(1) diff --git a/cwmscli/utils/__init__.py b/cwmscli/utils/__init__.py index 353538a..4b96df0 100644 --- a/cwmscli/utils/__init__.py +++ b/cwmscli/utils/__init__.py @@ -143,6 +143,18 @@ def log_scoped_read_hint( ) +def format_local_download_error(error: Exception, docs_url: str) -> str: + if isinstance(error, (OSError, ValueError)): + message = ( + f"Failed to download: {error}. " + f"If this is a local destination/path issue, pass --dest explicitly." + ) + if docs_url: + message = f"{message} Docs: {docs_url}" + return message + return f"Failed to download: {error}" + + def common_api_options(f): f = log_level_option(f) f = office_option(f) diff --git a/tests/commands/test_blob_upload.py b/tests/commands/test_blob_upload.py index 84fae93..050a8f7 100644 --- a/tests/commands/test_blob_upload.py +++ b/tests/commands/test_blob_upload.py @@ -1,5 +1,6 @@ import sys import types +import logging import pandas as pd import pytest @@ -312,6 +313,57 @@ class FakeHTTPError(Exception): assert calls == [("init_session", "https://example.test/", "apikey 123")] +def test_download_cmd_local_error_skips_scope_hint_and_logs_docs(monkeypatch, caplog): + class FakeBlobListing: + df = pd.DataFrame( + [{"id": "TEST_TXT", "media-type-id": "text/plain", "description": "x"}] + ) + + class FakeCwms: + @staticmethod + def init_session(api_root, api_key=None): + return None + + @staticmethod + def get_blob(office_id, blob_id): + return "retrieved text" + + @staticmethod + def get_blobs(office_id, blob_id_like): + return FakeBlobListing() + + monkeypatch.setitem(sys.modules, "cwms", FakeCwms) + + class FakeHTTPError(Exception): + pass + + monkeypatch.setitem( + sys.modules, "requests", types.SimpleNamespace(HTTPError=FakeHTTPError) + ) + monkeypatch.setattr( + "cwmscli.commands.blob.log_scoped_read_hint", + lambda **kwargs: (_ for _ in ()).throw(AssertionError("scope hint should not run")), + ) + monkeypatch.setattr( + "cwmscli.commands.blob._save_blob_content", + lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError("denied")), + ) + + with caplog.at_level(logging.ERROR), pytest.raises(SystemExit) as exc: + download_cmd( + blob_id="test_txt", + dest="downloaded", + office="SWT", + api_root="https://example.test/", + api_key="apikey 123", + dry_run=False, + ) + + assert exc.value.code == 1 + assert "pass --dest explicitly" in caplog.text + assert "/cli/blob.html" in caplog.text + + def test_list_cmd_initializes_session_with_api_key(monkeypatch): calls = [] diff --git a/tests/commands/test_clob.py b/tests/commands/test_clob.py index 727c143..9c09c96 100644 --- a/tests/commands/test_clob.py +++ b/tests/commands/test_clob.py @@ -1,7 +1,9 @@ import sys import types +import logging import pandas as pd +import pytest from cwmscli.commands import commands_cwms from cwmscli.commands.clob import ( @@ -248,6 +250,57 @@ class FakeHTTPError(Exception): assert calls == [("init_session", "https://example.test/", None)] + +def test_download_cmd_local_error_skips_scope_hint(monkeypatch, caplog): + class FakeClobResponse: + json = {"value": "retrieved clob text"} + + class FakeCwms: + @staticmethod + def init_session(api_root, api_key): + return None + + @staticmethod + def get_clob(office_id, clob_id): + return FakeClobResponse() + + monkeypatch.setitem(sys.modules, "cwms", FakeCwms) + monkeypatch.setattr("cwmscli.commands.clob.cwms", FakeCwms) + + class FakeHTTPError(Exception): + pass + + monkeypatch.setitem( + sys.modules, "requests", types.SimpleNamespace(HTTPError=FakeHTTPError) + ) + monkeypatch.setattr( + "cwmscli.commands.clob.requests", + types.SimpleNamespace(HTTPError=FakeHTTPError), + ) + monkeypatch.setattr( + "cwmscli.commands.clob.log_scoped_read_hint", + lambda **kwargs: (_ for _ in ()).throw(AssertionError("scope hint should not run")), + ) + monkeypatch.setattr( + "cwmscli.commands.clob._write_clob_content", + lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError("denied")), + ) + + with caplog.at_level(logging.ERROR), pytest.raises(SystemExit) as exc: + download_cmd( + clob_id="test_clob", + dest="downloaded.txt", + office="SWT", + api_root="https://example.test/", + api_key="apikey 123", + dry_run=False, + ) + + assert exc.value.code == 1 + assert "pass --dest explicitly" in caplog.text + assert "/cli/blob.html" not in caplog.text + + def test_list_cmd_initializes_session_with_api_key(monkeypatch): calls = [] From c9e750a79030795d932d89eb625c138f10fd27fc Mon Sep 17 00:00:00 2001 From: Charles Graham Date: Tue, 14 Apr 2026 12:50:48 -0500 Subject: [PATCH 4/7] Colorize issue 195 download path guidance --- cwmscli/utils/__init__.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/cwmscli/utils/__init__.py b/cwmscli/utils/__init__.py index 4b96df0..d3f2509 100644 --- a/cwmscli/utils/__init__.py +++ b/cwmscli/utils/__init__.py @@ -146,13 +146,17 @@ def log_scoped_read_hint( def format_local_download_error(error: Exception, docs_url: str) -> str: if isinstance(error, (OSError, ValueError)): message = ( - f"Failed to download: {error}. " - f"If this is a local destination/path issue, pass --dest explicitly." + f"{colors.c('Failed to download:', 'red', bright=True)} {error}. " + f"If this is a local destination/path issue, pass " + f"{colors.c('--dest', 'cyan', bright=True)} explicitly." ) if docs_url: - message = f"{message} Docs: {docs_url}" + message = ( + f"{message} {colors.c('Docs:', 'blue', bright=True)} " + f"{colors.c(docs_url, 'blue', bright=True)}" + ) return message - return f"Failed to download: {error}" + return f"{colors.c('Failed to download:', 'red', bright=True)} {error}" def common_api_options(f): From 1c55441ec97542cd0083c7bb9383bc46dade12e2 Mon Sep 17 00:00:00 2001 From: Charles Graham Date: Tue, 14 Apr 2026 13:38:36 -0500 Subject: [PATCH 5/7] Add exhaustive download destination safety tests --- tests/commands/test_clob.py | 12 ++--- tests/commands/test_download_dest_safety.py | 53 +++++++++++++++++++++ 2 files changed, 57 insertions(+), 8 deletions(-) create mode 100644 tests/commands/test_download_dest_safety.py diff --git a/tests/commands/test_clob.py b/tests/commands/test_clob.py index 9c09c96..eb07ecd 100644 --- a/tests/commands/test_clob.py +++ b/tests/commands/test_clob.py @@ -98,21 +98,17 @@ class FakeHTTPError(Exception): def test_download_cmd_default_dest_stays_relative_for_leading_slash_id( tmp_path, monkeypatch ): - class FakeClobResponse: - json = {"value": "retrieved clob text"} - class FakeCwms: @staticmethod def init_session(api_root, api_key): return None - @staticmethod - def get_clob(office_id, clob_id): - assert clob_id == "/REPORTS/REL-CLOB" - return FakeClobResponse() - monkeypatch.setitem(sys.modules, "cwms", FakeCwms) monkeypatch.setattr("cwmscli.commands.clob.cwms", FakeCwms) + monkeypatch.setattr( + "cwmscli.commands.clob._get_special_clob_text", + lambda office, clob_id: "retrieved clob text", + ) class FakeHTTPError(Exception): pass diff --git a/tests/commands/test_download_dest_safety.py b/tests/commands/test_download_dest_safety.py new file mode 100644 index 0000000..f95d7d9 --- /dev/null +++ b/tests/commands/test_download_dest_safety.py @@ -0,0 +1,53 @@ +import pytest + +from cwmscli.commands.blob import _default_download_dest as blob_default_download_dest +from cwmscli.commands.clob import _default_download_dest as clob_default_download_dest + + +SAFE_CASES = [ + ("REPORTS/REL-BLB", "REPORTS/REL-BLB"), + ("REPORTS\\REL-BLB", "REPORTS\\REL-BLB"), + ("/REPORTS/REL-BLB", "REPORTS/REL-BLB"), + ("\\REPORTS\\REL-BLB", "REPORTS\\REL-BLB"), +] + + +UNSAFE_CASES = [ + "", + "/", + "\\", + "../REPORTS/REL-BLB", + "..\\REPORTS\\REL-BLB", + "./REPORTS/REL-BLB", + ".\\REPORTS\\REL-BLB", + "REPORTS/../REL-BLB", + "REPORTS\\..\\REL-BLB", + "REPORTS/./REL-BLB", + "REPORTS\\.\\REL-BLB", + "C:/REPORTS/REL-BLB", + "C:\\REPORTS\\REL-BLB", + "//server/share/file", + "\\\\server\\share\\file", +] + + +@pytest.mark.parametrize("blob_id,expected", SAFE_CASES) +def test_blob_default_download_dest_allows_safe_relative_paths(blob_id, expected): + assert blob_default_download_dest(blob_id) == expected + + +@pytest.mark.parametrize("clob_id,expected", SAFE_CASES) +def test_clob_default_download_dest_allows_safe_relative_paths(clob_id, expected): + assert clob_default_download_dest(clob_id) == expected + + +@pytest.mark.parametrize("blob_id", UNSAFE_CASES) +def test_blob_default_download_dest_rejects_unsafe_paths(blob_id): + with pytest.raises(ValueError): + blob_default_download_dest(blob_id) + + +@pytest.mark.parametrize("clob_id", UNSAFE_CASES) +def test_clob_default_download_dest_rejects_unsafe_paths(clob_id): + with pytest.raises(ValueError): + clob_default_download_dest(clob_id) From 77cc22cf945934b85c4f914c8a6458a47a12242d Mon Sep 17 00:00:00 2001 From: Charles Graham Date: Tue, 14 Apr 2026 14:09:42 -0500 Subject: [PATCH 6/7] Fix unsafe default download destinations --- cwmscli/commands/blob.py | 13 +++--- cwmscli/commands/clob.py | 9 +---- cwmscli/utils/__init__.py | 45 +++++++++++++++++++++ tests/commands/test_blob_upload.py | 6 ++- tests/commands/test_clob.py | 6 ++- tests/commands/test_download_dest_safety.py | 1 - 6 files changed, 61 insertions(+), 19 deletions(-) diff --git a/cwmscli/commands/blob.py b/cwmscli/commands/blob.py index 781df0a..5697c56 100644 --- a/cwmscli/commands/blob.py +++ b/cwmscli/commands/blob.py @@ -14,6 +14,7 @@ format_local_download_error, get_api_key, log_scoped_read_hint, + validate_default_download_dest, ) from cwmscli.utils.click_help import DOCS_BASE_URL from cwmscli.utils.deps import requires @@ -109,13 +110,11 @@ def _save_blob_content( def _default_download_dest(blob_id: str) -> str: - target = blob_id.lstrip("/\\") - if not target: - raise ValueError( - f"Blob ID must include a non-root destination name. " - f"Pass --dest explicitly if needed. Docs: {BLOB_DOCS_URL}" - ) - return target + return validate_default_download_dest( + blob_id, + resource_name="Blob", + docs_url=BLOB_DOCS_URL, + ) def _blob_media_type(cwms_module, office: str, blob_id: str) -> Optional[str]: diff --git a/cwmscli/commands/clob.py b/cwmscli/commands/clob.py index 410bf0f..613467f 100644 --- a/cwmscli/commands/clob.py +++ b/cwmscli/commands/clob.py @@ -14,6 +14,7 @@ get_api_key, has_invalid_chars, log_scoped_read_hint, + validate_default_download_dest, ) @@ -35,13 +36,7 @@ def _write_clob_content(content: str, dest: str) -> str: def _default_download_dest(clob_id: str) -> str: - target = clob_id.lstrip("/\\") - if not target: - raise ValueError( - "Clob ID must include a non-root destination name. " - "Pass --dest explicitly if needed." - ) - return target + return validate_default_download_dest(clob_id, resource_name="Clob") def _clob_endpoint_id(clob_id: str) -> tuple[str, Optional[str]]: diff --git a/cwmscli/utils/__init__.py b/cwmscli/utils/__init__.py index d3f2509..18fdf6b 100644 --- a/cwmscli/utils/__init__.py +++ b/cwmscli/utils/__init__.py @@ -1,4 +1,5 @@ import logging as py_logging +import re from typing import Optional import click @@ -159,6 +160,50 @@ def format_local_download_error(error: Exception, docs_url: str) -> str: return f"{colors.c('Failed to download:', 'red', bright=True)} {error}" +def validate_default_download_dest( + raw_id: str, + *, + resource_name: str, + docs_url: str = "", +) -> str: + if raw_id is None: + raise ValueError( + f"{resource_name} ID must include a non-root destination name. " + f"Pass --dest explicitly if needed." + ) + + if raw_id.startswith("//") or raw_id.startswith("\\\\"): + raise ValueError( + f"{resource_name} ID must resolve to a relative local path. " + f"Pass --dest explicitly if needed." + ) + + target = raw_id.lstrip("/\\") + if not target: + message = ( + f"{resource_name} ID must include a non-root destination name. " + f"Pass --dest explicitly if needed." + ) + if docs_url: + message = f"{message} Docs: {docs_url}" + raise ValueError(message) + + if re.match(r"^[A-Za-z]:", target): + raise ValueError( + f"{resource_name} ID must resolve to a relative local path. " + f"Pass --dest explicitly if needed." + ) + + parts = re.split(r"[\\/]", target) + if any(part in {"", ".", ".."} for part in parts): + raise ValueError( + f"{resource_name} ID must resolve to a relative local path. " + f"Pass --dest explicitly if needed." + ) + + return target + + def common_api_options(f): f = log_level_option(f) f = office_option(f) diff --git a/tests/commands/test_blob_upload.py b/tests/commands/test_blob_upload.py index 050a8f7..e156658 100644 --- a/tests/commands/test_blob_upload.py +++ b/tests/commands/test_blob_upload.py @@ -1,6 +1,6 @@ +import logging import sys import types -import logging import pandas as pd import pytest @@ -342,7 +342,9 @@ class FakeHTTPError(Exception): ) monkeypatch.setattr( "cwmscli.commands.blob.log_scoped_read_hint", - lambda **kwargs: (_ for _ in ()).throw(AssertionError("scope hint should not run")), + lambda **kwargs: (_ for _ in ()).throw( + AssertionError("scope hint should not run") + ), ) monkeypatch.setattr( "cwmscli.commands.blob._save_blob_content", diff --git a/tests/commands/test_clob.py b/tests/commands/test_clob.py index eb07ecd..ad289c6 100644 --- a/tests/commands/test_clob.py +++ b/tests/commands/test_clob.py @@ -1,6 +1,6 @@ +import logging import sys import types -import logging import pandas as pd import pytest @@ -275,7 +275,9 @@ class FakeHTTPError(Exception): ) monkeypatch.setattr( "cwmscli.commands.clob.log_scoped_read_hint", - lambda **kwargs: (_ for _ in ()).throw(AssertionError("scope hint should not run")), + lambda **kwargs: (_ for _ in ()).throw( + AssertionError("scope hint should not run") + ), ) monkeypatch.setattr( "cwmscli.commands.clob._write_clob_content", diff --git a/tests/commands/test_download_dest_safety.py b/tests/commands/test_download_dest_safety.py index f95d7d9..a9f9d5c 100644 --- a/tests/commands/test_download_dest_safety.py +++ b/tests/commands/test_download_dest_safety.py @@ -3,7 +3,6 @@ from cwmscli.commands.blob import _default_download_dest as blob_default_download_dest from cwmscli.commands.clob import _default_download_dest as clob_default_download_dest - SAFE_CASES = [ ("REPORTS/REL-BLB", "REPORTS/REL-BLB"), ("REPORTS\\REL-BLB", "REPORTS\\REL-BLB"), From 7ca50dfa3af7688636a5d18132498b16302ec982 Mon Sep 17 00:00:00 2001 From: "Charles Graham, SWT" Date: Mon, 4 May 2026 22:34:19 -0500 Subject: [PATCH 7/7] Add monkeypatch type hints --- tests/commands/test_blob_upload.py | 42 ++++++++++++++++++++---------- tests/commands/test_clob.py | 34 ++++++++++++++++-------- 2 files changed, 51 insertions(+), 25 deletions(-) diff --git a/tests/commands/test_blob_upload.py b/tests/commands/test_blob_upload.py index 798f2c9..f9c037e 100644 --- a/tests/commands/test_blob_upload.py +++ b/tests/commands/test_blob_upload.py @@ -19,7 +19,7 @@ @pytest.fixture(autouse=True) -def no_saved_login(monkeypatch): +def no_saved_login(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr( "cwmscli.utils.get_saved_login_token", lambda *args, **kwargs: None ) @@ -46,7 +46,9 @@ def test_blob_id_for_path_uses_prefix_and_relative_path(): assert blob_id == "OPS_REPORTS_JAN_FINAL" -def test_upload_cmd_continues_on_error_for_directory(tmp_path, monkeypatch): +def test_upload_cmd_continues_on_error_for_directory( + tmp_path, monkeypatch: pytest.MonkeyPatch +): file_a = tmp_path / "a.txt" file_b = tmp_path / "b.txt" file_a.write_text("a") @@ -114,7 +116,7 @@ def test_find_blob_id_collisions_detects_same_stem_and_path_collisions(): def test_upload_cmd_aborts_before_upload_when_generated_ids_collide( - tmp_path, monkeypatch + tmp_path, monkeypatch: pytest.MonkeyPatch ): (tmp_path / "a.txt").write_text("a") (tmp_path / "a.json").write_text("{}") @@ -175,7 +177,9 @@ def test_default_download_dest_strips_leading_path_separators(): assert _default_download_dest("\\REPORTS\\REL-BLB") == "REPORTS\\REL-BLB" -def test_download_cmd_uses_media_type_to_write_text(tmp_path, monkeypatch): +def test_download_cmd_uses_media_type_to_write_text( + tmp_path, monkeypatch: pytest.MonkeyPatch +): dest = tmp_path / "downloaded" class FakeBlobListing: @@ -224,7 +228,7 @@ class FakeHTTPError(Exception): def test_download_cmd_default_dest_stays_relative_for_leading_slash_id( - tmp_path, monkeypatch + tmp_path, monkeypatch: pytest.MonkeyPatch ): class FakeBlobListing: df = pd.DataFrame( @@ -277,7 +281,9 @@ class FakeHTTPError(Exception): assert saved.read_text(encoding="utf-8") == "retrieved text" -def test_download_cmd_initializes_session_with_api_key(tmp_path, monkeypatch): +def test_download_cmd_initializes_session_with_api_key( + tmp_path, monkeypatch: pytest.MonkeyPatch +): dest = tmp_path / "downloaded" calls = [] @@ -321,7 +327,9 @@ class FakeHTTPError(Exception): assert calls == [("init_session", "https://example.test/", "apikey 123")] -def test_download_cmd_local_error_skips_scope_hint_and_logs_docs(monkeypatch, caplog): +def test_download_cmd_local_error_skips_scope_hint_and_logs_docs( + monkeypatch: pytest.MonkeyPatch, caplog +): class FakeBlobListing: df = pd.DataFrame( [{"id": "TEST_TXT", "media-type-id": "text/plain", "description": "x"}] @@ -374,7 +382,7 @@ class FakeHTTPError(Exception): assert "/cli/blob.html" in caplog.text -def test_list_cmd_initializes_session_with_api_key(monkeypatch): +def test_list_cmd_initializes_session_with_api_key(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -411,7 +419,7 @@ def get_blobs(office_id, blob_id_like, page_size=None): ] -def test_list_cmd_uses_limit_as_fetch_page_size(monkeypatch): +def test_list_cmd_uses_limit_as_fetch_page_size(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -447,7 +455,7 @@ def get_blobs(office_id, blob_id_like, page_size=None): assert calls == [("SWT", "TEST_.*", 25)] -def test_list_cmd_page_size_overrides_limit_for_fetch(monkeypatch): +def test_list_cmd_page_size_overrides_limit_for_fetch(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -480,7 +488,9 @@ def get_blobs(office_id, blob_id_like, page_size=None): assert calls == [("SWT", "TEST_.*", 200)] -def test_delete_cmd_uses_query_override_for_special_char_ids(monkeypatch): +def test_delete_cmd_uses_query_override_for_special_char_ids( + monkeypatch: pytest.MonkeyPatch, +): calls = [] class FakeApi: @@ -526,7 +536,9 @@ class FakeHTTPError(Exception): ] -def test_download_cmd_anonymous_skips_api_key(tmp_path, monkeypatch): +def test_download_cmd_anonymous_skips_api_key( + tmp_path, monkeypatch: pytest.MonkeyPatch +): dest = tmp_path / "downloaded" calls = [] @@ -571,7 +583,7 @@ class FakeHTTPError(Exception): assert calls == [("init_session", "https://example.test/", None)] -def test_list_cmd_anonymous_skips_api_key(monkeypatch): +def test_list_cmd_anonymous_skips_api_key(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -609,7 +621,9 @@ def get_blobs(office_id, blob_id_like, page_size=None): ] -def test_download_cmd_prefers_saved_token_over_api_key(tmp_path, monkeypatch): +def test_download_cmd_prefers_saved_token_over_api_key( + tmp_path, monkeypatch: pytest.MonkeyPatch +): dest = tmp_path / "downloaded" calls = [] diff --git a/tests/commands/test_clob.py b/tests/commands/test_clob.py index ad289c6..1f36d82 100644 --- a/tests/commands/test_clob.py +++ b/tests/commands/test_clob.py @@ -44,7 +44,9 @@ def test_default_download_dest_strips_leading_path_separators(): assert _default_download_dest("\\REPORTS\\REL-CLOB") == "REPORTS\\REL-CLOB" -def test_download_cmd_uses_default_dest_and_writes_text(tmp_path, monkeypatch): +def test_download_cmd_uses_default_dest_and_writes_text( + tmp_path, monkeypatch: pytest.MonkeyPatch +): calls = [] class FakeClobResponse: @@ -96,7 +98,7 @@ class FakeHTTPError(Exception): def test_download_cmd_default_dest_stays_relative_for_leading_slash_id( - tmp_path, monkeypatch + tmp_path, monkeypatch: pytest.MonkeyPatch ): class FakeCwms: @staticmethod @@ -137,7 +139,9 @@ class FakeHTTPError(Exception): assert saved.read_text(encoding="utf-8") == "retrieved clob text" -def test_download_cmd_uses_query_override_for_special_char_ids(tmp_path, monkeypatch): +def test_download_cmd_uses_query_override_for_special_char_ids( + tmp_path, monkeypatch: pytest.MonkeyPatch +): calls = [] class FakeCwms: @@ -204,7 +208,9 @@ class FakeHTTPError(Exception): ] -def test_download_cmd_anonymous_skips_api_key(tmp_path, monkeypatch): +def test_download_cmd_anonymous_skips_api_key( + tmp_path, monkeypatch: pytest.MonkeyPatch +): calls = [] class FakeClobResponse: @@ -247,7 +253,9 @@ class FakeHTTPError(Exception): assert calls == [("init_session", "https://example.test/", None)] -def test_download_cmd_local_error_skips_scope_hint(monkeypatch, caplog): +def test_download_cmd_local_error_skips_scope_hint( + monkeypatch: pytest.MonkeyPatch, caplog +): class FakeClobResponse: json = {"value": "retrieved clob text"} @@ -299,7 +307,7 @@ class FakeHTTPError(Exception): assert "/cli/blob.html" not in caplog.text -def test_list_cmd_initializes_session_with_api_key(monkeypatch): +def test_list_cmd_initializes_session_with_api_key(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -335,7 +343,7 @@ def get_clobs(office_id, clob_id_like, page_size=None): ] -def test_list_cmd_uses_limit_as_fetch_page_size(monkeypatch): +def test_list_cmd_uses_limit_as_fetch_page_size(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -372,7 +380,7 @@ def get_clobs(office_id, clob_id_like, page_size=None): assert calls == [("SWT", "TEST_.*", 25)] -def test_list_cmd_page_size_overrides_limit_for_fetch(monkeypatch): +def test_list_cmd_page_size_overrides_limit_for_fetch(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -404,7 +412,7 @@ def get_clobs(office_id, clob_id_like, page_size=None): assert calls == [("SWT", "TEST_.*", 200)] -def test_list_cmd_anonymous_skips_api_key(monkeypatch): +def test_list_cmd_anonymous_skips_api_key(monkeypatch: pytest.MonkeyPatch): calls = [] class FakeCwms: @@ -441,7 +449,9 @@ def get_clobs(office_id, clob_id_like, page_size=None): ] -def test_delete_cmd_uses_query_override_for_special_char_ids(monkeypatch): +def test_delete_cmd_uses_query_override_for_special_char_ids( + monkeypatch: pytest.MonkeyPatch, +): calls = [] class FakeCwms: @@ -481,7 +491,9 @@ def delete(endpoint, params=None): ] -def test_update_cmd_uses_query_override_for_special_char_ids(tmp_path, monkeypatch): +def test_update_cmd_uses_query_override_for_special_char_ids( + tmp_path, monkeypatch: pytest.MonkeyPatch +): calls = [] file_path = tmp_path / "updated.txt" file_path.write_text("updated clob text", encoding="utf-8")