#!/usr/bin/env python3
"""
A pull returns the unit of work: the issue AND what blocks it.

`--deps` used to be opt-in, so `pull.py 42` wrote a file with an empty
`depends:` and `issue_tree.py` drew it as a root with no blockers. The edge was
not lost — it lives in Gitea's native dependency graph — but it was not asked
for, and the body cannot supply it: `map.from_api` writes slugs into the
`## Depends on` prose and never `#N`. Following the graph is now the default.

What is asserted here:

1. **The default fills the graph.** A bare `pull.py <n>` fills `depends:` and
   pulls the blocker too, down to `--depth`.
2. **`--no-deps` is the way out, and it is free.** No `depends:`, no recursion,
   and not one request beyond the issue itself.
3. **`--deps` still works and means nothing.** Calls written against the old
   default keep running and get what they always got.
4. **The cost is one request per stored issue.** The native links are fetched
   once and used twice — for `depends:` and for the walk. Never twice.
5. **Filter mode follows blockers out of the selection, deliberately.** A
   blocker no filter selected still lands in the store and does not spend
   `--limit`; a closed one is dropped like any other closed issue, and so is
   the edge to it. An issue the filter dropped costs no link request at all.

The transport is stubbed at `_gitea.api`, as the other suites do it, and the
stub records every call so "how many requests" is an observation. No network,
and no test touches the developer's store: each builds its own in a
`tempfile.TemporaryDirectory()`.
"""
import contextlib
import io
import os
import sys
import tempfile
import unittest
import urllib.parse
from unittest import mock

_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
for _p in (os.path.join(_ROOT, "skills", "sync", "scripts"),
           os.path.join(_ROOT, "skills", "issue", "scripts")):
    if _p not in sys.path:
        sys.path.insert(0, _p)

import _gitea  # noqa: E402
import issue  # noqa: E402
import pull  # noqa: E402

REPO = "claude-skills/tea"
BASE = "repos/%s" % REPO

BODY = """## Summary
Прозаическое описание задачи.

## Spec
skills/issue/references/format.md

## Acceptance criteria
- [ ] что-нибудь работает
"""


def payload(number, title, state="open"):
    return {"number": number, "title": title, "body": BODY, "state": state,
            "comments": 0, "labels": [{"name": "type/task"}], "assignees": [],
            "milestone": None, "ref": "main", "updated_at": "2026-08-10T00:00:00Z",
            "html_url": "https://git.example/%s/issues/%d" % (REPO, number),
            "repository": {"full_name": REPO}}


class FakeTracker(object):
    """`tea api` answered from memory, with a native dependency graph.

    `listed` is what the list endpoint serves — the filter's selection. `extra`
    exists and is fetchable by number but is in no selection, which is how a
    blocker outside the filter is modelled. `deps` maps a blocked issue's number
    to the numbers that block it, the direction `GET …/dependencies` reads.
    """

    def __init__(self, listed=(), extra=(), deps=None):
        self.listed = list(listed)
        self.issues = {p["number"]: p for p in list(listed) + list(extra)}
        self.deps = {int(k): list(v) for k, v in (deps or {}).items()}
        self.calls = []        # (method, path), in request order

    # -- what the tests read off it ----------------------------------------

    def paths(self, suffix):
        return [p for m, p in self.calls if p.endswith(suffix)]

    def issue_gets(self):
        """`GET …/issues/<n>` — one issue fetched by number."""
        return [p for m, p in self.calls
                if m == "GET" and p.startswith("%s/issues/" % BASE)
                and p.rsplit("/", 1)[1].isdigit()]

    # -- the seam ----------------------------------------------------------

    def api(self, login, endpoint, method="GET", payload=None,
            payload_name=None, out_root=None, allow_fail=False):
        path, _, qs = endpoint.partition("?")
        q = urllib.parse.parse_qs(qs)
        self.calls.append((method, path))

        if path == "%s/issues" % BASE and method == "GET":
            page, per = int(q["page"][0]), int(q["limit"][0])
            return self.listed[(page - 1) * per:(page - 1) * per + per]

        if path.endswith("/comments"):
            return []

        if path.endswith("/dependencies") and method == "GET":
            n = int(path.split("/issues/")[1].split("/")[0])
            return [self.issues[b] for b in self.deps.get(n, []) if b in self.issues]

        if path.startswith("%s/issues/" % BASE) and method == "GET":
            return self.issues.get(int(path.rsplit("/", 1)[1]))

        raise AssertionError("unstubbed call: %s %s" % (method, endpoint))


class PullDepsTestCase(unittest.TestCase):
    """A temp store, a fake tracker, no git and no network."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix="tea-deps-")
        self.addCleanup(self.tmp.cleanup)
        self.root = os.path.join(self.tmp.name, "tmp", "issues")
        os.makedirs(self.root)
        p = mock.patch.object(_gitea, "require_login", lambda: "test-login")
        p.start()
        self.addCleanup(p.stop)

    def serve(self, listed=(), extra=(), deps=None):
        self.fake = FakeTracker(listed, extra, deps)
        p = mock.patch.object(_gitea, "api", self.fake.api)
        p.start()
        self.addCleanup(p.stop)
        return self.fake

    def blocked_pair(self):
        """#10 "Second thing" is blocked by #7 "First thing"."""
        return self.serve(listed=[payload(10, "Second thing"),
                                  payload(7, "First thing")],
                          deps={10: [7]})

    def run_pull(self, *argv):
        out, err = io.StringIO(), io.StringIO()
        args = ["pull.py", "--repo", REPO, "--out", self.root] + list(argv)
        with mock.patch.object(sys, "argv", args), \
                contextlib.redirect_stdout(out), \
                contextlib.redirect_stderr(err):
            pull.main()
        return out.getvalue(), err.getvalue()

    def stored(self):
        return sorted(issue.all_ids(self.root))

    def depends_of(self, id):
        return issue.load(self.root, id).depends


# --------------------------------------------------------------------------
# 1. the default fills the graph
# --------------------------------------------------------------------------

class DepsAreTheDefaultTest(PullDepsTestCase):

    def test_a_bare_pull_fills_depends(self):
        """The acceptance criterion, and the whole point: no flag, and the file
        knows what blocks it."""
        self.blocked_pair()
        self.run_pull("10")
        self.assertEqual(self.depends_of("second-thing"), ["first-thing"])

    def test_a_bare_pull_stores_the_blocker(self):
        """`depends:` pointing at a file that is not there would be worse than
        an empty one — the blocker comes with it."""
        self.blocked_pair()
        self.run_pull("10")
        self.assertIn("first-thing", self.stored())

    def test_the_walk_is_recursive(self):
        """A blocker's blocker is context too, down to --depth (default 3)."""
        self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 6)],
                   deps={1: [2], 2: [3], 3: [4], 4: [5]})
        self.run_pull("1")
        self.assertEqual(self.stored(), ["thing-1", "thing-2", "thing-3", "thing-4"],
                         "the default depth of 3 was not what was walked")

    def test_depth_bounds_the_walk(self):
        self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 6)],
                   deps={1: [2], 2: [3], 3: [4], 4: [5]})
        self.run_pull("1", "--depth", "1")
        self.assertEqual(self.stored(), ["thing-1", "thing-2"])

    def test_the_graph_hint_is_printed_when_there_is_a_graph(self):
        self.blocked_pair()
        out, _ = self.run_pull("10")
        self.assertIn("issue_tree.py", out)


# --------------------------------------------------------------------------
# 2. --no-deps is the way out, and it is free
# --------------------------------------------------------------------------

class NoDepsOptsOutTest(PullDepsTestCase):

    def test_no_deps_leaves_depends_empty(self):
        self.blocked_pair()
        self.run_pull("10", "--no-deps")
        self.assertEqual(self.depends_of("second-thing"), [])

    def test_no_deps_does_not_pull_the_blocker(self):
        self.blocked_pair()
        self.run_pull("10", "--no-deps")
        self.assertEqual(self.stored(), ["second-thing"])

    def test_no_deps_spends_no_extra_request(self):
        """The other half of the criterion: not the links, not the blocker.
        One issue asked for, one request made."""
        self.blocked_pair()
        self.run_pull("10", "--no-deps")
        self.assertEqual(self.fake.paths("/dependencies"), [])
        self.assertEqual(self.fake.issue_gets(), ["%s/issues/10" % BASE])

    def test_no_deps_prints_no_graph_hint(self):
        self.blocked_pair()
        out, _ = self.run_pull("10", "--no-deps")
        self.assertNotIn("issue_tree.py", out)


# --------------------------------------------------------------------------
# 3. --deps is still accepted, and means nothing
# --------------------------------------------------------------------------

class DepsFlagIsANoOpTest(PullDepsTestCase):

    def test_the_flag_is_still_accepted(self):
        """Existing calls and the /tea:sync command tables must not break."""
        self.blocked_pair()
        self.run_pull("10", "--deps")
        self.assertEqual(self.depends_of("second-thing"), ["first-thing"])

    def test_it_changes_nothing_about_the_run(self):
        self.blocked_pair()
        self.run_pull("10", "--deps")
        with_flag = (self.stored(), self.depends_of("second-thing"),
                     list(self.fake.calls))

        self.setUp()
        self.blocked_pair()
        self.run_pull("10")
        self.assertEqual((self.stored(), self.depends_of("second-thing"),
                          list(self.fake.calls)), with_flag)


# --------------------------------------------------------------------------
# 4. one request per stored issue
# --------------------------------------------------------------------------

class TheCostIsOneRequestPerIssueTest(PullDepsTestCase):

    def test_the_links_are_fetched_once_per_issue(self):
        """They fill `depends:` AND steer the walk; fetching them twice is
        double the price the docstring quotes."""
        self.blocked_pair()
        self.run_pull("10")
        self.assertEqual(self.fake.paths("/dependencies"),
                         ["%s/issues/10/dependencies" % BASE,
                          "%s/issues/7/dependencies" % BASE])

    def test_a_bulk_pull_costs_one_per_issue(self):
        """The number the docstring quotes: one list request, then one link
        request per issue that lands in the store."""
        self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 21)])
        self.run_pull("-q", "x")
        self.assertEqual(len(self.fake.paths("/dependencies")), 20)
        self.assertEqual(len(self.fake.paths("/issues")), 1)

    def test_a_cached_issue_costs_its_links_and_nothing_else(self):
        """--cached stops the body and the thread, not the graph: a cached
        issue's blockers can be missing from disk even when it is not."""
        self.blocked_pair()
        self.run_pull("10", "--no-deps")          # only #10 on disk
        self.fake.calls = []
        self.run_pull("10", "--cached")
        self.assertEqual(self.fake.paths("/dependencies"),
                         ["%s/issues/10/dependencies" % BASE,
                          "%s/issues/7/dependencies" % BASE])
        self.assertIn("first-thing", self.stored())


# --------------------------------------------------------------------------
# 5. filter mode follows blockers out of the selection
# --------------------------------------------------------------------------

class FilterModeFollowsOutwardTest(PullDepsTestCase):

    def test_a_blocker_outside_the_filter_lands_in_the_store(self):
        """Documented as deliberate: a blocker is followed because a stored
        issue named it, not because the filter selected it."""
        self.serve(listed=[payload(1, "Selected thing")],
                   extra=[payload(99, "Outside thing")],
                   deps={1: [99]})
        self.run_pull("-q", "x")
        self.assertEqual(self.stored(), ["outside-thing", "selected-thing"])
        self.assertEqual(self.depends_of("selected-thing"), ["outside-thing"])

    def test_a_blocker_does_not_spend_the_limit(self):
        """--limit counts the selection's writes; the graph is not part of the
        selection, so the store can legitimately hold more than N."""
        self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 5)],
                   extra=[payload(100 + n, "Blocker %d" % n) for n in range(1, 5)],
                   deps={n: [100 + n] for n in range(1, 5)})
        self.run_pull("-q", "x", "--limit", "2")
        self.assertEqual(self.stored(),
                         ["blocker-1", "blocker-2", "thing-1", "thing-2"])

    def test_a_closed_blocker_is_dropped_with_the_edge_to_it(self):
        """The documented exception. Closed is not a unit of work, so filter
        mode drops it like any other closed issue — and `depends:` must not be
        left pointing at a file that is not there."""
        self.serve(listed=[payload(1, "Selected thing")],
                   extra=[payload(99, "Closed blocker", state="closed")],
                   deps={1: [99]})
        self.run_pull("-q", "x")
        self.assertEqual(self.stored(), ["selected-thing"])
        self.assertEqual(self.depends_of("selected-thing"), [])

    def test_a_closed_blocker_is_stored_in_key_mode(self):
        """An address is not a bulk read: `pull.py 1` has no closed rule."""
        self.serve(listed=[payload(1, "Selected thing")],
                   extra=[payload(99, "Closed blocker", state="closed")],
                   deps={1: [99]})
        self.run_pull("1")
        self.assertEqual(self.stored(), ["closed-blocker", "selected-thing"])

    def test_a_dropped_closed_issue_costs_no_link_request(self):
        """Nothing was stored for it, so there is no unit of work to complete
        — and its own blockers are not dragged in behind it."""
        self.serve(listed=[payload(1, "Closed thing", state="closed"),
                           payload(2, "Open thing")],
                   extra=[payload(50, "Blocker of the closed one")],
                   deps={1: [50]})
        self.run_pull("-q", "x", "--state", "all")
        self.assertEqual(self.fake.paths("/dependencies"),
                         ["%s/issues/2/dependencies" % BASE])
        self.assertEqual(self.stored(), ["open-thing"])


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