import json import os import tempfile import unittest from argparse import Namespace from pathlib import Path from unittest.mock import patch import refurbished_watcher as watcher LISTING_HTML = """

Refurbished Mac Studio Apple M2 Max Chip with 12-Core CPU and 30-Core GPU, 64GB unified memory, 1TB SSD

$1,999.00 """ EMBEDDED_LISTING_HTML = """ """ MAC_MINI_CRITERIA_HTML = """ """ HIGH_MEMORY_LISTING_HTML = """ """ def restore_env(key, value): if value is None: os.environ.pop(key, None) else: os.environ[key] = value class RefurbishedWatcherTests(unittest.TestCase): def test_parse_listing_filters_targets_and_prices(self): # Only Mac Studio products with at least 64GB RAM match. products = watcher.parse_listing(LISTING_HTML, watcher.DEFAULT_URL) self.assertEqual([product.key for product in products], ["G2STUDIO"]) self.assertEqual(products[0].price, "$1,999.00") def test_parse_embedded_tiles_ignores_non_studio_products(self): products = watcher.parse_listing(EMBEDDED_LISTING_HTML, watcher.DEFAULT_URL) self.assertEqual(products, []) def test_parse_embedded_tiles_matches_high_memory_mac_studios_only(self): products = watcher.parse_listing(HIGH_MEMORY_LISTING_HTML, watcher.DEFAULT_URL) self.assertEqual([product.key for product in products], ["G2AAA/A", "G2BBB/A"]) self.assertEqual(products[0].ram, "96GB") self.assertEqual(products[1].ram, "128GB") def test_filter_targets_enriches_candidates_with_missing_ram(self): product = watcher.Product( key="G2TARGET", title="Refurbished Mac Studio Apple M2 Ultra Chip", price="$3,999.00", url="https://www.apple.com/shop/product/G2TARGET/refurbished-mac-studio", ) with patch.object(watcher, "enrich_product", return_value=watcher.Product(**{**product.__dict__, "ram": "96GB", "storage": "1TB"})): products = watcher.filter_targets([product], watcher.DEFAULT_CRITERIA, timeout=1) self.assertEqual([item.key for item in products], ["G2TARGET"]) def test_mac_mini_does_not_match(self): products = watcher.parse_listing(MAC_MINI_CRITERIA_HTML, watcher.DEFAULT_URL) self.assertEqual(products, []) def test_extract_details(self): text = "Configurable specs include 24GB unified memory and 512GB SSD storage." self.assertEqual(watcher.extract_ram(text), "24GB") self.assertEqual(watcher.extract_storage(text), "512GB") def test_canonical_product_url_removes_volatile_query(self): url = "https://www.apple.com/shop/product/g2aaa/a/refurbished-mac-studio?fnode=abc#section" self.assertEqual( watcher.canonical_product_url(url), "https://www.apple.com/shop/product/g2aaa/a/refurbished-mac-studio", ) def test_first_run_baselines_without_notifications(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(Path(tmpdir) / "inventory.jsonl"), inventory_state_file=str(Path(tmpdir) / "inventory-state.json"), timeout=1, target_title_contains=watcher.DEFAULT_TARGET_TITLE_CONTAINS, min_ram_gb=watcher.MAC_STUDIO_MIN_RAM_GB, max_products=0, notify_on_first_run=False, dry_run=False, no_notify=False, add_to_cart=False, ntfy_server="https://ntfy.sh", topic="refurbishedMac", ) with patch.object(watcher, "fetch_text", return_value=LISTING_HTML), patch.object(watcher, "notify_ntfy") as notify: new_products = watcher.check_once(args) self.assertEqual(new_products, []) notify.assert_not_called() state = json.loads(state_file.read_text()) self.assertEqual(state["seen_keys"], ["G2STUDIO"]) def test_new_product_after_previous_check_notifies(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" state_file.write_text(json.dumps({"seen_keys": ["OLDMATCH"]})) args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(Path(tmpdir) / "inventory.jsonl"), inventory_state_file=str(Path(tmpdir) / "inventory-state.json"), timeout=1, target_title_contains=watcher.DEFAULT_TARGET_TITLE_CONTAINS, min_ram_gb=watcher.MAC_STUDIO_MIN_RAM_GB, max_products=0, notify_on_first_run=False, dry_run=False, no_notify=False, add_to_cart=False, ntfy_server="https://ntfy.sh", topic="refurbishedMac", ) with ( patch.object(watcher, "fetch_text", return_value=LISTING_HTML), patch.object(watcher, "enrich_product", side_effect=lambda product, timeout=1: product), patch.object(watcher, "notify_ntfy") as notify, ): new_products = watcher.check_once(args) self.assertEqual([product.key for product in new_products], ["G2STUDIO"]) notify.assert_called_once() state = json.loads(state_file.read_text()) self.assertEqual(state["seen_keys"], ["G2STUDIO"]) def test_new_product_after_empty_baseline_notifies(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" state_file.write_text(json.dumps({"seen_keys": []})) args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(Path(tmpdir) / "inventory.jsonl"), inventory_state_file=str(Path(tmpdir) / "inventory-state.json"), timeout=1, target_title_contains=watcher.DEFAULT_TARGET_TITLE_CONTAINS, min_ram_gb=watcher.MAC_STUDIO_MIN_RAM_GB, max_products=0, notify_on_first_run=False, dry_run=False, no_notify=False, add_to_cart=False, ntfy_server="https://ntfy.sh", topic="refurbishedMac", ) with ( patch.object(watcher, "fetch_text", return_value=LISTING_HTML), patch.object(watcher, "enrich_product", side_effect=lambda product, timeout=1: product), patch.object(watcher, "notify_ntfy") as notify, ): new_products = watcher.check_once(args) self.assertEqual([product.key for product in new_products], ["G2STUDIO"]) notify.assert_called_once() def test_new_product_can_add_to_cart_before_notify(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" state_file.write_text(json.dumps({"seen_keys": ["OLDMATCH"]})) args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(Path(tmpdir) / "inventory.jsonl"), inventory_state_file=str(Path(tmpdir) / "inventory-state.json"), timeout=1, target_title_contains=watcher.DEFAULT_TARGET_TITLE_CONTAINS, min_ram_gb=watcher.MAC_STUDIO_MIN_RAM_GB, max_products=0, notify_on_first_run=False, dry_run=False, no_notify=False, add_to_cart=True, cart_profile=str(Path(tmpdir) / "cart-profile"), cart_headless=False, cart_keep_open=False, ntfy_server="https://ntfy.sh", topic="refurbishedMac", ) with ( patch.object(watcher, "fetch_text", return_value=LISTING_HTML), patch.object(watcher, "enrich_product", side_effect=lambda product, timeout=1: product), patch.object(watcher, "add_products_to_cart", return_value=[]) as add_to_cart, patch.object(watcher, "notify_ntfy") as notify, ): new_products = watcher.check_once(args) self.assertEqual([product.key for product in new_products], ["G2STUDIO"]) add_to_cart.assert_called_once() notify.assert_called_once() def test_cart_failure_still_notifies(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" state_file.write_text(json.dumps({"seen_keys": ["OLDMATCH"]})) args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(Path(tmpdir) / "inventory.jsonl"), inventory_state_file=str(Path(tmpdir) / "inventory-state.json"), timeout=1, target_title_contains=watcher.DEFAULT_TARGET_TITLE_CONTAINS, min_ram_gb=watcher.MAC_STUDIO_MIN_RAM_GB, max_products=0, notify_on_first_run=False, dry_run=False, no_notify=False, add_to_cart=True, cart_profile=str(Path(tmpdir) / "cart-profile"), cart_headless=False, cart_keep_open=False, ntfy_server="https://ntfy.sh", topic="refurbishedMac", ) with ( patch.object(watcher, "fetch_text", return_value=LISTING_HTML), patch.object(watcher, "enrich_product", side_effect=lambda product, timeout=1: product), patch.object(watcher, "add_products_to_cart", side_effect=RuntimeError("missing browser")), patch.object(watcher, "notify_ntfy") as notify, ): new_products = watcher.check_once(args) self.assertEqual([product.key for product in new_products], ["G2STUDIO"]) notify.assert_called_once() def test_report_cart_setup_runs_only_when_cart_enabled(self): args = Namespace(add_to_cart=False, cart_profile="unused", timeout=1) with patch.object(watcher, "cart_setup_errors") as setup_errors: watcher.report_cart_setup(args) setup_errors.assert_not_called() def test_report_cart_setup_can_disable_cart_when_user_continues(self): args = Namespace(add_to_cart=True, cart_profile="unused", timeout=1) status = watcher.CartSetupStatus(ok=False, code="launch_failed", message="cannot launch") with ( patch.object(watcher, "cart_setup_status", return_value=status), patch.object(watcher, "prompt_yes_no", return_value=True), ): watcher.report_cart_setup(args) self.assertFalse(args.add_to_cart) def test_report_cart_setup_exits_when_user_declines_notifier_only(self): args = Namespace(add_to_cart=True, cart_profile="unused", timeout=1) status = watcher.CartSetupStatus(ok=False, code="launch_failed", message="cannot launch") with ( patch.object(watcher, "cart_setup_status", return_value=status), patch.object(watcher, "prompt_yes_no", return_value=False), ): with self.assertRaises(SystemExit): watcher.report_cart_setup(args) def test_report_cart_setup_installs_missing_chromium(self): args = Namespace(add_to_cart=True, cart_profile="unused", timeout=1) missing = watcher.CartSetupStatus(ok=False, code="browser_missing", message="Executable doesn't exist") ok = watcher.CartSetupStatus(ok=True, code="ok", message="Playwright can launch Chromium.") with ( patch.object(watcher, "cart_setup_status", side_effect=[missing, ok]), patch.object(watcher, "prompt_yes_no", return_value=True), patch.object(watcher, "install_playwright_chromium", return_value=True) as install, ): watcher.report_cart_setup(args) install.assert_called_once() self.assertTrue(args.add_to_cart) def test_criteria_can_be_overridden_for_cart_testing(self): criteria = watcher.Criteria(title_contains="mac mini", min_ram_gb=24) products = watcher.parse_listing(MAC_MINI_CRITERIA_HTML, watcher.DEFAULT_URL, criteria) self.assertEqual([product.key for product in products], ["G4EEE/A", "G4AAA/A", "G4BBB/A", "G4DDD/A"]) def test_max_products_limits_handled_new_products(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" state_file.write_text(json.dumps({"seen_keys": ["OLDMATCH"]})) args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(Path(tmpdir) / "inventory.jsonl"), inventory_state_file=str(Path(tmpdir) / "inventory-state.json"), timeout=1, target_title_contains="mac studio", min_ram_gb=64, max_products=1, notify_on_first_run=False, dry_run=True, no_notify=False, add_to_cart=False, ntfy_server="https://ntfy.sh", topic="refurbishedMac", ) with ( patch.object(watcher, "fetch_text", return_value=HIGH_MEMORY_LISTING_HTML), patch.object(watcher, "enrich_product", side_effect=lambda product, timeout=1: product), ): new_products = watcher.check_once(args) self.assertEqual([product.key for product in new_products], ["G2AAA/A"]) def test_record_inventory_snapshot_logs_only_changes(self): with tempfile.TemporaryDirectory() as tmpdir: history_path = Path(tmpdir) / "history.jsonl" state_path = Path(tmpdir) / "inventory-state.json" products = [ watcher.Product(key="A", title="Refurbished Mac mini", ram="16GB"), watcher.Product(key="B", title="Refurbished Mac Studio", ram="96GB"), ] first = watcher.record_inventory_snapshot(products, products[1:], history_path, state_path) second = watcher.record_inventory_snapshot(products, products[1:], history_path, state_path) self.assertTrue(first.changed) self.assertFalse(second.changed) lines = history_path.read_text().splitlines() self.assertEqual(len(lines), 1) event = json.loads(lines[0]) self.assertEqual(event["total_products"], 2) self.assertEqual(event["matched_keys"], ["B"]) def test_record_poll_snapshot_logs_every_poll_compactly(self): with tempfile.TemporaryDirectory() as tmpdir: poll_history_path = Path(tmpdir) / "polls.jsonl" products = [ watcher.Product(key="A", title="Refurbished Mac mini", ram="16GB"), watcher.Product(key="B", title="Refurbished Mac Studio", ram="96GB"), ] snapshot = watcher.InventorySnapshot( fingerprint="abc123", products=products, matched_keys=["B"], added_keys=["A"], removed_keys=[], changed=True, ) watcher.record_poll_snapshot(snapshot, poll_history_path) lines = poll_history_path.read_text().splitlines() self.assertEqual(len(lines), 1) event = json.loads(lines[0]) self.assertEqual(event["fingerprint"], "abc123") self.assertTrue(event["changed"]) self.assertEqual(event["total_products"], 2) self.assertEqual(event["matched_count"], 1) self.assertEqual(event["added_count"], 1) self.assertEqual(event["removed_count"], 0) self.assertNotIn("products", event) def test_check_once_records_poll_history_even_when_inventory_is_unchanged(self): with tempfile.TemporaryDirectory() as tmpdir: state_file = Path(tmpdir) / "state.json" inventory_history_file = Path(tmpdir) / "inventory.jsonl" inventory_state_file = Path(tmpdir) / "inventory-state.json" poll_history_file = Path(tmpdir) / "polls.jsonl" args = Namespace( url=watcher.DEFAULT_URL, state_file=str(state_file), inventory_history_file=str(inventory_history_file), inventory_state_file=str(inventory_state_file), poll_history_file=str(poll_history_file), timeout=1, target_title_contains=watcher.DEFAULT_TARGET_TITLE_CONTAINS, min_ram_gb=watcher.MAC_STUDIO_MIN_RAM_GB, max_products=0, notify_on_first_run=False, dry_run=False, no_notify=True, add_to_cart=False, ntfy_server="https://ntfy.sh", topic="", ) with patch.object(watcher, "fetch_text", return_value=LISTING_HTML): watcher.check_once(args) watcher.check_once(args) poll_events = [json.loads(line) for line in poll_history_file.read_text().splitlines()] inventory_events = inventory_history_file.read_text().splitlines() self.assertEqual(len(poll_events), 2) self.assertEqual(len(inventory_events), 1) self.assertTrue(poll_events[0]["changed"]) self.assertFalse(poll_events[1]["changed"]) def test_load_env_file_sets_missing_values_without_overriding_environment(self): with tempfile.TemporaryDirectory() as tmpdir: env_file = Path(tmpdir) / ".env.local" env_file.write_text( "\n".join( [ "NTFY_TOPIC=from-file", "TARGET_TITLE_CONTAINS='mac studio'", "MIN_RAM_GB=64", ] ) ) old_topic = os.environ.get("NTFY_TOPIC") old_title = os.environ.get("TARGET_TITLE_CONTAINS") old_ram = os.environ.get("MIN_RAM_GB") try: os.environ["NTFY_TOPIC"] = "from-env" os.environ.pop("TARGET_TITLE_CONTAINS", None) os.environ.pop("MIN_RAM_GB", None) watcher.load_env_file(str(env_file)) self.assertEqual(os.environ["NTFY_TOPIC"], "from-env") self.assertEqual(os.environ["TARGET_TITLE_CONTAINS"], "mac studio") self.assertEqual(os.environ["MIN_RAM_GB"], "64") finally: restore_env("NTFY_TOPIC", old_topic) restore_env("TARGET_TITLE_CONTAINS", old_title) restore_env("MIN_RAM_GB", old_ram) def test_validate_args_requires_topic_for_notification_runs(self): args = Namespace(dry_run=False, no_notify=False, topic="") with self.assertRaises(SystemExit): watcher.validate_args(args) def test_validate_args_allows_missing_topic_when_notifications_disabled(self): watcher.validate_args(Namespace(dry_run=True, no_notify=False, topic="")) watcher.validate_args(Namespace(dry_run=False, no_notify=True, topic="")) if __name__ == "__main__": unittest.main()

Refurbished Mac mini Apple M4 Chip with 10-Core CPU and 10-Core GPU, Gigabit Ethernet

Refurbished 13-inch MacBook Air Apple M4 chip with 10-Core CPU and 8-Core GPU - Sky Blue

Refurbished 15-inch MacBook Air Apple M4 chip with 10-Core CPU and 10-Core GPU - Silver

Refurbished Mac Studio Apple M2 Max Chip with 12-Core CPU and 30-Core GPU, 64GB unified memory, 1TB SSD