mirror of
https://github.com/seanbetts/steam-hardware-watch.git
synced 2026-10-06 04:04:22 +02:00
454 lines
17 KiB
Python
454 lines
17 KiB
Python
import csv
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
IMPORTINFO_HTML = textwrap.dedent(
|
|
"""
|
|
<html>
|
|
<body>
|
|
<h2>Search Results</h2>
|
|
<table>
|
|
<thead>
|
|
<tr>
|
|
<th>Run Date</th><th>Master BOL</th><th>House BOL</th><th>Voyage #</th>
|
|
<th>Bill Type</th><th>Carrier Code</th><th>IMO #</th><th>Vessel Name</th>
|
|
<th>Arrival Date</th><th>US Port</th><th>Foreign Port</th><th>Quantity</th>
|
|
<th>Weight</th><th>Type of Service</th><th>Shipper</th><th>Consignee</th>
|
|
<th>Notify Party</th><th>Commodity</th>
|
|
</tr>
|
|
</thead>
|
|
<tbody>
|
|
<tr>
|
|
<td>2026-05-01</td><td>EGLV142653125618</td><td>SNHBSHALAX264014</td><td>084E</td>
|
|
<td>House Bill</td><td>SNHB</td><td>9604081</td><td>EVER LOGIC</td>
|
|
<td>2026-05-01</td><td>LOS ANGELES, CALIFORNIA</td><td>SHANGHAI CHINA (MAINLAND)</td>
|
|
<td>42 PKG</td><td>12,578 K</td><td>House to House</td>
|
|
<td>TECH-FRONT (CHONGQING) COMPUTER CO</td><td>CEVA C/O VALVE CORPORATION</td>
|
|
<td>CEVA C/O VALVE CORPORATION</td><td>GAME CONSOLE</td>
|
|
</tr>
|
|
<tr>
|
|
<td>2026-04-28</td><td>CMDUCHN3170474</td><td>EXDO621128151</td><td>0XRAJ</td>
|
|
<td>House Bill</td><td>EXDO</td><td>9436379</td><td>CMA CGM SAMSON</td>
|
|
<td>2026-04-28</td><td>SAVANNAH, GEORGIA</td><td>SHANGHAI CHINA (MAINLAND)</td>
|
|
<td>15 PKG</td><td>4,143 KG</td><td>Pier to Pier</td>
|
|
<td>TECH-FRONT (CHONGQING) COMPUTER CO</td><td>PROMETHEAN INC.</td>
|
|
<td></td><td>CHROMEBOX HTS:</td>
|
|
</tr>
|
|
</tbody>
|
|
</table>
|
|
</body>
|
|
</html>
|
|
"""
|
|
)
|
|
|
|
|
|
def run_parser(args, reports):
|
|
return subprocess.run(
|
|
[
|
|
"python3",
|
|
str(ROOT / "scripts" / "parse_importinfo_shipments.py"),
|
|
*args,
|
|
"--report-dir",
|
|
str(reports),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def read_rows(reports):
|
|
return list(
|
|
csv.DictReader(
|
|
(reports / "customs-shipments.tsv").read_text(encoding="utf-8").splitlines(),
|
|
delimiter="\t",
|
|
)
|
|
)
|
|
|
|
|
|
def write_executable(path, content):
|
|
path.write_text(textwrap.dedent(content).lstrip(), encoding="utf-8")
|
|
path.chmod(path.stat().st_mode | stat.S_IXUSR)
|
|
|
|
|
|
class CustomsShipmentParserTests(unittest.TestCase):
|
|
def test_extracts_relevant_importinfo_rows(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
html = tmp_path / "ceva-valve.html"
|
|
html.write_text(IMPORTINFO_HTML, encoding="utf-8")
|
|
reports = tmp_path / "reports"
|
|
|
|
result = run_parser(["--input", f"ceva-valve={html}"], reports)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
|
|
rows = read_rows(reports)
|
|
self.assertEqual(1, len(rows))
|
|
self.assertEqual("SNHBSHALAX264014", rows[0]["house_bol"])
|
|
self.assertEqual("GAME CONSOLE", rows[0]["commodity"])
|
|
self.assertEqual("CEVA C/O VALVE CORPORATION", rows[0]["consignee"])
|
|
|
|
key_lines = (reports / "customs-shipments-key-lines.txt").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("2026-05-01", key_lines)
|
|
self.assertIn("SNHBSHALAX264014", key_lines)
|
|
self.assertIn("GAME CONSOLE", key_lines)
|
|
|
|
report = (reports / "customs-shipments.md").read_text(encoding="utf-8")
|
|
self.assertIn("## Newest Relevant Shipments", report)
|
|
self.assertIn("CEVA C/O VALVE CORPORATION", report)
|
|
self.assertIn("TECH-FRONT (CHONGQING) COMPUTER CO", report)
|
|
|
|
def test_preserves_distinct_same_day_bols(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
html = tmp_path / "same-day.html"
|
|
html.write_text(
|
|
IMPORTINFO_HTML.replace(
|
|
"</tbody>",
|
|
textwrap.dedent(
|
|
"""
|
|
<tr>
|
|
<td>2026-05-01</td><td>EGLV142653125669</td><td>SNHBSHALAX264015</td><td>084E</td>
|
|
<td>House Bill</td><td>SNHB</td><td>9604081</td><td>EVER LOGIC</td>
|
|
<td>2026-05-01</td><td>LOS ANGELES, CALIFORNIA</td><td>SHANGHAI CHINA (MAINLAND)</td>
|
|
<td>42 PKG</td><td>12,596 K</td><td>House to House</td>
|
|
<td>TECH-FRONT (CHONGQING) COMPUTER CO</td><td>CEVA C/O VALVE CORPORATION</td>
|
|
<td>CEVA C/O VALVE CORPORATION</td><td>GAME CONSOLE</td>
|
|
</tr>
|
|
</tbody>
|
|
"""
|
|
),
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
reports = tmp_path / "reports"
|
|
|
|
result = run_parser(["--input", f"ceva-valve={html}"], reports)
|
|
|
|
self.assertEqual(0, result.returncode)
|
|
rows = read_rows(reports)
|
|
self.assertEqual(
|
|
["SNHBSHALAX264014", "SNHBSHALAX264015"],
|
|
[row["house_bol"] for row in rows],
|
|
)
|
|
|
|
def test_ignores_later_unrelated_table_headers(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
html = tmp_path / "extra-table.html"
|
|
html.write_text(
|
|
IMPORTINFO_HTML.replace(
|
|
"</body>",
|
|
textwrap.dedent(
|
|
"""
|
|
<table>
|
|
<tr><th>Name</th><th>Description</th></tr>
|
|
<tr><td>PROMETHEAN INC.</td><td>CHROMEBOX HTS:</td></tr>
|
|
</table>
|
|
</body>
|
|
"""
|
|
),
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
reports = tmp_path / "reports"
|
|
|
|
result = run_parser(["--input", f"ceva-valve={html}"], reports)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
rows = read_rows(reports)
|
|
self.assertEqual(["SNHBSHALAX264014"], [row["house_bol"] for row in rows])
|
|
|
|
def test_matches_relevance_terms_split_by_inline_markup(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
html = tmp_path / "split-cells.html"
|
|
html.write_text(
|
|
IMPORTINFO_HTML.replace(
|
|
"CEVA C/O VALVE CORPORATION</td><td>GAME CONSOLE",
|
|
"CEVA C/O <span>VALVE</span> CORPORATION</td><td>GAME<br>CONSOLE",
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
reports = tmp_path / "reports"
|
|
|
|
result = run_parser(["--input", f"ceva-valve={html}"], reports)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
rows = read_rows(reports)
|
|
self.assertEqual(1, len(rows))
|
|
self.assertEqual("CEVA C/O VALVE CORPORATION", rows[0]["consignee"])
|
|
self.assertEqual("GAME CONSOLE", rows[0]["commodity"])
|
|
|
|
def test_filters_generic_logistics_product_rows_without_valve_signal(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
html = tmp_path / "generic-logistics.html"
|
|
html.write_text(
|
|
IMPORTINFO_HTML.replace("CEVA C/O VALVE CORPORATION", "CEVA LOGISTICS US INC."),
|
|
encoding="utf-8",
|
|
)
|
|
reports = tmp_path / "reports"
|
|
|
|
result = run_parser(["--input", f"tech-front-game-console={html}"], reports)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
self.assertEqual([], read_rows(reports))
|
|
|
|
def test_manifest_inputs_dedupe_duplicate_bols(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
first = tmp_path / "first.html"
|
|
second = tmp_path / "second.html"
|
|
manifest = tmp_path / "manifest.tsv"
|
|
first.write_text(IMPORTINFO_HTML, encoding="utf-8")
|
|
second.write_text(IMPORTINFO_HTML, encoding="utf-8")
|
|
manifest.write_text(
|
|
"\n".join(
|
|
(
|
|
"slug\tsource_url\thtml_path",
|
|
f"ceva-valve\thttps://example.test/ceva\t{first}",
|
|
f"ceva-valve\thttps://example.test/ceva\t{second}",
|
|
)
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
reports = tmp_path / "reports"
|
|
|
|
result = run_parser(["--manifest", str(manifest)], reports)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
rows = read_rows(reports)
|
|
self.assertEqual(["SNHBSHALAX264014"], [row["house_bol"] for row in rows])
|
|
self.assertEqual("https://example.test/ceva", rows[0]["source_url"])
|
|
|
|
|
|
class CheckCustomsShipmentsShellTests(unittest.TestCase):
|
|
def test_fetches_pages_and_runs_parser(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
run_dir = tmp_path / "run"
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
write_executable(
|
|
fake_bin / "curl",
|
|
f"""
|
|
#!/bin/sh
|
|
cat <<'HTML'
|
|
{IMPORTINFO_HTML}
|
|
HTML
|
|
""",
|
|
)
|
|
env = os.environ.copy()
|
|
env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}"
|
|
|
|
result = subprocess.run(
|
|
[str(ROOT / "scripts" / "check_customs_shipments.sh"), str(run_dir)],
|
|
cwd=ROOT,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
self.assertTrue(
|
|
(run_dir / "api" / "customs" / "importinfo-ceva-valve.html").exists()
|
|
)
|
|
key_lines = (
|
|
run_dir / "reports" / "customs-shipments-key-lines.txt"
|
|
).read_text(encoding="utf-8")
|
|
self.assertIn("SNHBSHALAX264014", key_lines)
|
|
|
|
def test_records_blocked_fetches_without_failing_run(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
run_dir = tmp_path / "run"
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
write_executable(
|
|
fake_bin / "curl",
|
|
"""
|
|
#!/bin/sh
|
|
exit 22
|
|
""",
|
|
)
|
|
env = os.environ.copy()
|
|
env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}"
|
|
|
|
result = subprocess.run(
|
|
[str(ROOT / "scripts" / "check_customs_shipments.sh"), str(run_dir)],
|
|
cwd=ROOT,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
errors = (run_dir / "reports" / "customs-shipments-errors.txt").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("failed to fetch", errors)
|
|
|
|
def test_records_200_challenge_page_without_key_lines(self):
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
run_dir = tmp_path / "run"
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
write_executable(
|
|
fake_bin / "curl",
|
|
"""
|
|
#!/bin/sh
|
|
cat <<'HTML'
|
|
<html><body><h1>Checking your browser</h1><p>Please verify you are human.</p></body></html>
|
|
HTML
|
|
""",
|
|
)
|
|
env = os.environ.copy()
|
|
env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}"
|
|
|
|
result = subprocess.run(
|
|
[str(ROOT / "scripts" / "check_customs_shipments.sh"), str(run_dir)],
|
|
cwd=ROOT,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual("", result.stderr)
|
|
self.assertEqual(0, result.returncode)
|
|
errors = (run_dir / "reports" / "customs-shipments-errors.txt").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("unexpected or blocked content from", errors)
|
|
self.assertFalse(
|
|
(run_dir / "api" / "customs" / "importinfo-ceva-valve.html").exists()
|
|
)
|
|
key_lines = (
|
|
run_dir / "reports" / "customs-shipments-key-lines.txt"
|
|
).read_text(encoding="utf-8")
|
|
self.assertEqual("", key_lines)
|
|
|
|
|
|
class CustomsSummaryIntegrationTests(unittest.TestCase):
|
|
def test_run_summary_and_status_draft_include_customs_lines(self):
|
|
with TemporaryDirectory() as tmp:
|
|
run_dir = Path(tmp) / "2026-05-05"
|
|
reports = run_dir / "reports"
|
|
reports.mkdir(parents=True)
|
|
(reports / "customs-shipments-key-lines.txt").write_text(
|
|
"2026-05-01\tCEVA C/O VALVE CORPORATION\tTECH-FRONT (CHONGQING) COMPUTER CO\tGAME CONSOLE\t42 PKG\t12596 Kgs\tSNHBSHALAX264015\n",
|
|
encoding="utf-8",
|
|
)
|
|
(reports / "customs-shipments-errors.txt").write_text("", encoding="utf-8")
|
|
|
|
summary = subprocess.run(
|
|
[
|
|
"python3",
|
|
str(ROOT / "scripts" / "write_run_summary.py"),
|
|
"--run-dir",
|
|
str(run_dir),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
draft = subprocess.run(
|
|
[
|
|
"python3",
|
|
str(ROOT / "scripts" / "draft_status_update.py"),
|
|
"--run-dir",
|
|
str(run_dir),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(0, summary.returncode)
|
|
self.assertEqual(0, draft.returncode)
|
|
run_summary = (reports / "run-summary.md").read_text(encoding="utf-8")
|
|
status_draft = (reports / "status-draft.md").read_text(encoding="utf-8")
|
|
self.assertIn("Customs shipments", run_summary)
|
|
self.assertIn("SNHBSHALAX264015", run_summary)
|
|
self.assertIn("Customs / Shipments", status_draft)
|
|
self.assertIn("GAME CONSOLE", status_draft)
|
|
|
|
def test_status_draft_reports_missing_customs_outputs_as_unavailable(self):
|
|
with TemporaryDirectory() as tmp:
|
|
run_dir = Path(tmp) / "2026-05-05"
|
|
reports = run_dir / "reports"
|
|
reports.mkdir(parents=True)
|
|
|
|
draft = subprocess.run(
|
|
[
|
|
"python3",
|
|
str(ROOT / "scripts" / "draft_status_update.py"),
|
|
"--run-dir",
|
|
str(run_dir),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
summary = subprocess.run(
|
|
[
|
|
"python3",
|
|
str(ROOT / "scripts" / "write_run_summary.py"),
|
|
"--run-dir",
|
|
str(run_dir),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(0, draft.returncode)
|
|
self.assertEqual(0, summary.returncode)
|
|
status_draft = (reports / "status-draft.md").read_text(encoding="utf-8")
|
|
run_summary = (reports / "run-summary.md").read_text(encoding="utf-8")
|
|
self.assertIn("Customs / Shipments", status_draft)
|
|
self.assertIn("not generated", status_draft)
|
|
self.assertNotIn("No relevant customs shipment rows found.", status_draft)
|
|
self.assertIn("Customs shipments status: `unavailable`", run_summary)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|