test_search_posts.py

"""Offline tutorial checks. All responses and credentials below are synthetic."""

import contextlib
import io
import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qs, urlsplit

import search_posts


def response(ids, charged, token=None):
    meta = {"request_id": "synthetic-request", "credits": {"charged": charged, "remaining": 100}}
    if token is not None:
        meta["pagination"] = {"next_token": token}
    return io.BytesIO(json.dumps({
        "data": [{"id": post_id, "text": "Synthetic example: café #AI"} for post_id in ids],
        "meta": meta,
    }).encode())


class SearchTutorialTests(unittest.TestCase):
    def test_pagination_dedup_encoding_and_confirmed_costs(self):
        query = '(#AI OR $BTC) "agent tools" lang:en'
        token = "opaque+/=token?one&two"
        with patch.object(search_posts, "urlopen", side_effect=[
            response(["9007199254740993", "2"], 3, token), response(["2", "3"], 3)
        ]) as opened:
            result = search_posts.collect_posts("synthetic-key", query, 3)
        self.assertEqual([post["id"] for post in result["data"]], ["9007199254740993", "2", "3"])
        self.assertEqual(result["credits_charged_confirmed"], 6)
        self.assertEqual(result["stop_reason"], "no_next_token")
        first, second = [call.args[0] for call in opened.call_args_list]
        self.assertEqual(urlsplit(first.full_url).hostname, "api.xfetch.io")
        self.assertEqual(urlsplit(first.full_url).fragment, "")
        first_params = parse_qs(urlsplit(first.full_url).query)
        second_params = parse_qs(urlsplit(second.full_url).query)
        self.assertEqual(first_params, {"query": [query], "limit": ["20"], "sort_order": ["recency"]})
        self.assertEqual(second_params.pop("next_token"), [token])
        self.assertEqual(first_params, second_params)
        self.assertEqual(first.get_header("Authorization"), "Bearer synthetic-key")
        self.assertTrue(all(call.kwargs["timeout"] == 30 for call in opened.call_args_list))

    def test_empty_page_with_forward_token_continues(self):
        with patch.object(search_posts, "urlopen", side_effect=[
            response([], 1, "next"), response(["1"], 2)
        ]) as opened:
            result = search_posts.collect_posts("synthetic-key", "AI", 3)
        self.assertEqual(opened.call_count, 2)
        self.assertEqual(result["credits_charged_confirmed"], 3)
        self.assertEqual(len(result["data"]), 1)

    def test_page_limit_is_reported_without_claiming_exhaustion(self):
        with patch.object(search_posts, "urlopen", side_effect=[response(["1"], 2, "more")]) as opened:
            result = search_posts.collect_posts("synthetic-key", "AI", 1)
        self.assertEqual(opened.call_count, 1)
        self.assertEqual(result["stop_reason"], "page_limit")

    def test_repeated_token_stops_without_a_third_request(self):
        with patch.object(search_posts, "urlopen", side_effect=[
            response(["1"], 2, "same"), response(["1"], 2, "same")
        ]) as opened:
            result = search_posts.collect_posts("synthetic-key", "AI", 3)
        self.assertEqual(opened.call_count, 2)
        self.assertEqual(result["stop_reason"], "repeated_token")
        self.assertEqual(result["credits_charged_confirmed"], 4)

    def test_http_failure_retains_prior_posts_without_retry(self):
        error = HTTPError(search_posts.ENDPOINT, 429, "synthetic", {}, None)
        with patch.object(search_posts, "urlopen", side_effect=[response(["1"], 2, "next"), error]) as opened:
            result = search_posts.collect_posts("synthetic-key", "AI", 3)
        self.assertEqual(opened.call_count, 2)
        self.assertEqual(result["pages_fetched"], 1)
        self.assertEqual(result["credits_charged_confirmed"], 2)
        self.assertEqual(result["stop_reason"], "error")
        self.assertEqual(result["data"][0]["id"], "1")
        self.assertIn("HTTP 429", result["error"])

    def test_network_failure_does_not_claim_zero_total_cost(self):
        with patch.object(search_posts, "urlopen", side_effect=URLError("synthetic")):
            result = search_posts.collect_posts("synthetic-key", "AI", 3)
        self.assertEqual(result["credits_charged_confirmed"], 0)
        self.assertIn("charge is unknown", result["error"])

    def test_malformed_response_is_an_error(self):
        for body in [b"not json", b'{"data":[]}', b'{"data":[],"meta":{"credits":{"charged":false}}}']:
            with self.subTest(body=body), patch.object(search_posts, "urlopen", return_value=io.BytesIO(body)):
                result = search_posts.collect_posts("synthetic-key", "AI", 3)
                self.assertEqual(result["stop_reason"], "error")

    def test_cli_saves_partial_json_and_returns_failure(self):
        with tempfile.TemporaryDirectory() as folder:
            output = Path(folder) / "partial.json"
            with patch.dict(os.environ, {"XFETCH_API_KEY": "synthetic-key"}), \
                 patch.object(search_posts, "urlopen", side_effect=[response(["1"], 2, "next"), TimeoutError()]), \
                 contextlib.redirect_stderr(io.StringIO()):
                code = search_posts.main(["AI", "--output", str(output)])
            data = json.loads(output.read_text())
            self.assertEqual(code, 1)
            self.assertEqual(data["stop_reason"], "error")
            self.assertEqual(data["data"][0]["text"], "Synthetic example: café #AI")
            self.assertNotIn("synthetic-key", output.read_text())

    def test_existing_file_and_missing_key_fail_before_network(self):
        with tempfile.TemporaryDirectory() as folder, patch.object(search_posts, "urlopen") as opened:
            output = Path(folder) / "existing.json"
            output.write_text("keep me")
            with patch.dict(os.environ, {"XFETCH_API_KEY": "synthetic-key"}), contextlib.redirect_stderr(io.StringIO()):
                with self.assertRaises(SystemExit):
                    search_posts.main(["AI", "--output", str(output)])
            with patch.dict(os.environ, {"XFETCH_API_KEY": ""}), contextlib.redirect_stderr(io.StringIO()):
                with self.assertRaises(SystemExit):
                    search_posts.main(["AI"])
            opened.assert_not_called()
            self.assertEqual(output.read_text(), "keep me")

    def test_invalid_page_budget_fails_before_network(self):
        with patch.object(search_posts, "urlopen") as opened, contextlib.redirect_stderr(io.StringIO()):
            for count in ["0", "4"]:
                with self.assertRaises(SystemExit):
                    search_posts.main(["AI", "--max-pages", count])
            opened.assert_not_called()


if __name__ == "__main__":
    unittest.main()