skills/magnus919_agent-skills
1
0
Fork 0
mirror of https://github.com/magnus919/agent-skills.git synced 2026-09-11 19:47:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(raleigh): skip token-gated imagery folders in discovery and canary (#280)

Browse Source
This commit is contained in:
Magnus Hedemark
2026-08-05 11:16:33 -04:00
committed by GitHub
parent 3eb7bd4096
commit e99c6d2994
5 changed files with 176 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
raleigh/SKILL.md
+5
View File
@@ -65,6 +65,11 @@ scripts/raleigh download "Raleigh Dog Parks" -f csv -o dog_parks.csv
| `imagery identify` | Identify pixel value at point | `scripts/raleigh imagery identify Orthos2025 --point=-78.65,35.75` |
| `imagery statistics` | Compute extent statistics | `scripts/raleigh imagery statistics Orthos2025 --bbox=-78.7,35.7,-78.6,35.8` |
`imagery catalog` lists only publicly readable services. Folders whose
listing requires a token (currently `Imagery` and `Utilities`) are skipped
and reported as restricted rather than failing the command; the daily live
canary tracks them the same way.
### Geocoding
| Command | Purpose | Example |
raleigh/scripts/canary.py
+48 -7
View File
@@ -6,8 +6,13 @@ non-Hub adapter endpoint shipped by the CLI. Validates source-specific
minimum schemas, classifies failures, retries bounded transient errors,
and writes a machine-readable JSON report.
Token-gated imagery folders are reported as ``restricted_folder``
observations (non-failing): the CLI only reads public data, so a folder
that requires a token is not a contract failure, but it stays visible in
the report.
Exit codes:
0 all probes passed (or only empty-but-valid observations)
0 all probes passed (or only empty-but-valid / restricted observations)
1 one or more durable contract failures detected
2 script-level error (bad arguments, import failure, etc.)
"""
@@ -48,6 +53,7 @@ FAILURE_CLASSES = (
"schema_drift",
"parser_failure",
"empty_but_valid",
"restricted_folder",
)
@@ -346,15 +352,25 @@ def probe_meetings() -> list[dict[str, Any]]:
def probe_imagery_catalog() -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
services, err = _probe_with_retry(imagery.list_services)
listing, err = _probe_with_retry(imagery.list_services)
if err:
results.append({"source": "imagery", "target": "catalog", "status": "fail", **err})
return results
if not isinstance(services, list):
if not isinstance(listing, tuple) or len(listing) != 2:
results.append({
"source": "imagery", "target": "catalog", "status": "fail",
"failure_class": "schema_drift", "error": "expected list of services", "attempt": 1,
"failure_class": "schema_drift",
"error": "list_services returned an unexpected shape", "attempt": 1,
})
return results
services, restricted_folders = listing
if not isinstance(services, list) or not isinstance(restricted_folders, list):
results.append({
"source": "imagery", "target": "catalog", "status": "fail",
"failure_class": "schema_drift",
"error": "list_services returned non-list fields", "attempt": 1,
})
return results
@@ -363,9 +379,18 @@ def probe_imagery_catalog() -> list[dict[str, Any]]:
"source": "imagery", "target": "catalog", "status": "pass",
"failure_class": "empty_but_valid", "error": "no imagery services", "attempt": 1,
})
return results
else:
results.append({"source": "imagery", "target": "catalog", "status": "pass", "count": len(services)})
for folder in restricted_folders:
results.append({
"source": "imagery",
"target": f"folder:{folder}",
"status": "pass",
"failure_class": "restricted_folder",
"error": "folder listing requires a token; skipped",
"attempt": 1,
})
return results
@@ -455,6 +480,7 @@ def run_canary() -> dict[str, Any]:
durable_failures = 0
transient_failures = 0
empty_valid = 0
restricted = 0
for name, probe_fn in ALL_PROBES:
try:
@@ -477,6 +503,8 @@ def run_canary() -> dict[str, Any]:
if r.get("status") != "fail":
if r.get("failure_class") == "empty_but_valid":
empty_valid += 1
elif r.get("failure_class") == "restricted_folder":
restricted += 1
continue
fc = r.get("failure_class", "unknown")
if _is_transient(fc):
@@ -495,6 +523,7 @@ def run_canary() -> dict[str, Any]:
"durable_failures": durable_failures,
"transient_failures": transient_failures,
"empty_but_valid": empty_valid,
"restricted_folders": restricted,
},
"results": all_results,
}
@@ -516,6 +545,17 @@ def write_github_summary(report: dict[str, Any]) -> None:
lines.append(f"| Durable failures | {s['durable_failures']} |")
lines.append(f"| Transient failures | {s['transient_failures']} |")
lines.append(f"| Empty-but-valid | {s['empty_but_valid']} |")
lines.append(f"| Restricted folders | {s.get('restricted_folders', 0)} |")
lines.append("")
restricted = [r for r in report["results"] if r.get("failure_class") == "restricted_folder"]
if restricted:
lines.append("### Restricted folders (token required; skipped, non-failing)")
lines.append("")
lines.append("| Source | Target | Evidence |")
lines.append("|--------|--------|----------|")
for r in restricted:
lines.append(f"| {r.get('source', '?')} | {r.get('target', '?')} | {r.get('error', '')[:120]} |")
lines.append("")
failures = [r for r in report["results"] if r.get("status") == "fail"]
@@ -551,7 +591,8 @@ def main() -> int:
f"{s['total_results']} probes, "
f"{s['durable_failures']} durable failures, "
f"{s['transient_failures']} transient failures, "
f"{s['empty_but_valid']} empty-but-valid"
f"{s['empty_but_valid']} empty-but-valid, "
f"{s.get('restricted_folders', 0)} restricted folders"
)
print(f"Report written to {report_path}")
raleigh/scripts/raleighlib/cli.py
+12 -1
View File
@@ -683,12 +683,23 @@ def cmd_categories(args: argparse.Namespace) -> int:
def cmd_imagery_catalog(args: argparse.Namespace) -> int:
services = imagery.list_services()
services, restricted_folders = imagery.list_services()
limit = args.limit
if args.json:
_output_json(services[:limit])
if restricted_folders:
print(
f"note: {len(restricted_folders)} imagery folder(s) require a token "
f"and were skipped: {', '.join(restricted_folders)}",
file=sys.stderr,
)
else:
_output_table(["NAME", "TYPE"], [[s.get("name", ""), s.get("type", "")] for s in services[:limit]])
if restricted_folders:
print(
f"Note: {len(restricted_folders)} imagery folder(s) require a token "
f"and were skipped: {', '.join(restricted_folders)}"
)
return 0
raleigh/scripts/raleighlib/imagery.py
+21 -3
View File
@@ -24,12 +24,23 @@ def _checked_json(data: Any, operation: str) -> dict[str, Any]:
return data
def _is_token_required_error(exc: Exception) -> bool:
"""Return True when an ArcGIS error indicates the resource needs a token."""
return "token required" in str(exc).lower()
def list_services(
root_url: str = IMAGE_ROOT,
max_folders: int = MAX_IMAGE_FOLDERS,
max_services: int = MAX_IMAGE_SERVICES,
) -> list[dict[str, Any]]:
"""Recursively discover ImageServer services from the REST directory."""
) -> tuple[list[dict[str, Any]], list[str]]:
"""Recursively discover ImageServer services from the REST directory.
Returns ``(services, restricted_folders)`` where ``restricted_folders``
names folders whose listing requires a token. Those folders are skipped:
this tool only reads publicly accessible services, and a token-gated folder
is not a public-data contract violation.
"""
if max_folders < 0 or max_services < 1:
raise ValueError("imagery discovery bounds are invalid")
sep = "&" if "?" in root_url else "?"
@@ -55,12 +66,19 @@ def list_services(
f"Image service listing exceeded {max_services} services"
)
services = list(root_services)
restricted_folders: list[str] = []
for folder in folders:
folder_url = f"{root_url}/{urllib.parse.quote(folder, safe='')}"
sep = "&" if "?" in folder_url else "?"
try:
folder_data = _checked_json(
core.json_request(f"{folder_url}{sep}f=pjson"), "Image folder listing"
)
except ValueError as exc:
if _is_token_required_error(exc):
restricted_folders.append(folder)
continue
raise
folder_services = folder_data.get("services", [])
if not isinstance(folder_services, list) or any(
not isinstance(service, dict) for service in folder_services
@@ -74,7 +92,7 @@ def list_services(
svc = dict(svc)
svc["folder"] = folder
services.append(svc)
return services
return services, restricted_folders
def service_info(url: str) -> dict[str, Any]:
raleigh/tests/test_raleigh.py
+87 -3
View File
@@ -452,10 +452,45 @@ class ImageryTests(unittest.TestCase):
return root
with patch("raleighlib.imagery.core.json_request", side_effect=mock_response):
services = imagery.list_services()
services, restricted = imagery.list_services()
names = {s["name"] for s in services}
self.assertIn("Base/Image", names)
self.assertIn("Ortho/2025", names)
self.assertEqual(restricted, [])
def test_list_services_skips_token_required_folder(self):
root = {
"folders": ["Public", "Gated"],
"services": [],
}
public_folder = {"services": [{"name": "Public/Ortho", "type": "ImageServer"}]}
def mock_response(url):
if "Gated" in url:
raise ValueError("Image folder listing failed: Token Required")
if "Public" in url:
return public_folder
return root
with patch("raleighlib.imagery.core.json_request", side_effect=mock_response):
services, restricted = imagery.list_services()
self.assertEqual([s["name"] for s in services], ["Public/Ortho"])
self.assertEqual(restricted, ["Gated"])
def test_list_services_still_raises_on_non_token_folder_error(self):
root = {
"folders": ["Broken"],
"services": [],
}
def mock_response(url):
if "Broken" in url:
raise ValueError("Image folder listing failed: Invalid URL")
return root
with patch("raleighlib.imagery.core.json_request", side_effect=mock_response):
with self.assertRaisesRegex(ValueError, "Invalid URL"):
imagery.list_services()
def test_supports_capability(self):
info = {"capabilities": "Catalog,Image,Metadata"}
@@ -1236,7 +1271,7 @@ class CliTests(unittest.TestCase):
def test_imagery_catalog_subcommand(self):
with patch("raleighlib.cli.imagery.list_services") as mock_list:
mock_list.return_value = [{"name": "Orthos2025", "type": "ImageServer"}]
mock_list.return_value = ([{"name": "Orthos2025", "type": "ImageServer"}], [])
code, out, err = self.run_cli(["imagery", "catalog", "--json"])
self.assertEqual(code, 0, err)
self.assertIn("Orthos2025", out)
@@ -1560,11 +1595,24 @@ class GlobalFlagTests(unittest.TestCase):
def test_global_flags_after_nested_subcommand(self):
with patch("raleighlib.cli.imagery.list_services") as mock_list:
mock_list.return_value = [{"name": "Orthos2025", "type": "ImageServer"}]
mock_list.return_value = ([{"name": "Orthos2025", "type": "ImageServer"}], [])
code, out, err = self.run_cli(["imagery", "catalog", "--json", "--limit", "1"])
self.assertEqual(code, 0, err)
self.assertIn("Orthos2025", out)
def test_imagery_catalog_reports_restricted_folders(self):
with patch("raleighlib.cli.imagery.list_services") as mock_list:
mock_list.return_value = (
[{"name": "Orthos2025", "type": "ImageServer"}],
["Imagery", "Utilities"],
)
code, out, err = self.run_cli(["imagery", "catalog"])
self.assertEqual(code, 0, err)
self.assertIn("Orthos2025", out)
self.assertIn("Imagery", out)
self.assertIn("Utilities", out)
self.assertIn("require a token", out)
def test_global_flags_before_subcommand(self):
with patch("raleighlib.cli.hub.catalog_from_cache_or_live") as mock_catalog:
mock_catalog.return_value = [{"id": "x", "title": "Test", "type": "FeatureServer", "access": "public"}]
@@ -2678,6 +2726,42 @@ class PoliceTests(unittest.TestCase):
self.assertFalse(report["passed"])
self.assertEqual(report["summary"]["transient_failures"], 1)
def test_canary_imagery_probe_reports_restricted_folders_as_non_failing(self):
with patch("canary.imagery.list_services", return_value=(
[{"name": "Orthos2025", "type": "ImageServer"}],
["Imagery", "Utilities"],
)):
results = canary_lib.probe_imagery_catalog()
self.assertTrue(all(result["status"] == "pass" for result in results))
targets = {result["target"]: result.get("failure_class") for result in results}
self.assertEqual(targets["catalog"], None)
self.assertEqual(targets["folder:Imagery"], "restricted_folder")
self.assertEqual(targets["folder:Utilities"], "restricted_folder")
self.assertEqual(len(results), 3)
def test_canary_imagery_probe_fails_on_unexpected_shape(self):
with patch("canary.imagery.list_services", return_value=[
{"name": "Orthos2025", "type": "ImageServer"},
]):
results = canary_lib.probe_imagery_catalog()
self.assertEqual(len(results), 1)
self.assertEqual(results[0]["status"], "fail")
self.assertEqual(results[0]["failure_class"], "schema_drift")
def test_canary_summary_counts_restricted_folders_without_failing(self):
observations = [
{"source": "imagery", "target": "catalog", "status": "pass", "count": 1},
{
"source": "imagery", "target": "folder:Imagery", "status": "pass",
"failure_class": "restricted_folder", "error": "token required", "attempt": 1,
},
]
with patch.object(canary_lib, "ALL_PROBES", [("imagery", lambda: observations)]):
report = canary_lib.run_canary()
self.assertTrue(report["passed"])
self.assertEqual(report["summary"]["restricted_folders"], 1)
self.assertEqual(report["summary"]["durable_failures"], 0)
class FireTests(unittest.TestCase):
"""Tests for the RFD incident command group and 2026 schema normalization."""