path: root/tests/test_subcmds_status.py

# Copyright (C) 2026 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unittests for the status subcmd."""

import contextlib
import io
import os
from pathlib import Path
import subprocess
from typing import List, Tuple
from unittest import mock

import pytest
import utils_for_test

import manifest_xml
import subcmds


@pytest.fixture
def repo_client_checkout(
    tmp_path: Path,
) -> Tuple[Path, manifest_xml.XmlManifest]:
    """Create a basic repo client checkout for status tests."""
    # Create in a subdir to avoid noise (like the repo_trace file).
    topdir = tmp_path / "client_checkout"
    repodir = topdir / ".repo"
    manifest_dir = repodir / "manifests"
    manifest_file = repodir / manifest_xml.MANIFEST_FILE_NAME

    repodir.mkdir(parents=True)
    manifest_dir.mkdir()

    gitdir = repodir / "manifests.git"
    gitdir.mkdir()
    (gitdir / "config").write_text(
        """[remote "origin"]
                url = https://localhost:0/manifest
                verbose = false
            """
    )

    _init_temp_git_tree(manifest_dir)

    manifest_file.write_text(
        """
            <manifest>
                <remote name="origin" fetch="http://localhost" />
                <default remote="origin" revision="refs/heads/main" />
                <project name="proj" path="src/proj" />
            </manifest>
        """,
        encoding="utf-8",
    )

    (repodir / "projects" / "src" / "proj.git").mkdir(parents=True)
    (repodir / "project-objects" / "proj.git").mkdir(parents=True)

    worktree = topdir / "src" / "proj"
    worktree.parent.mkdir(parents=True, exist_ok=True)
    _init_temp_git_tree(worktree)

    manifest = manifest_xml.XmlManifest(str(repodir), str(manifest_file))
    return topdir, manifest


def _init_temp_git_tree(git_dir: Path) -> None:
    """Create a new git checkout with an initial commit for testing."""
    utils_for_test.init_git_tree(git_dir)
    (git_dir / "README").write_text("init")
    subprocess.check_call(["git", "add", "README"], cwd=git_dir)
    subprocess.check_call(["git", "commit", "-q", "-m", "init"], cwd=git_dir)


def _run_status(manifest: manifest_xml.XmlManifest, argv: List[str]) -> None:
    """Run the status subcommand with parsed options against a test manifest."""
    cmd = subcmds.status.Status()
    cmd.manifest = manifest
    cmd.client = mock.MagicMock(globalConfig=manifest.globalConfig)

    opts, args = cmd.OptionParser.parse_args(argv + ["--jobs=1"])
    cmd.CommonValidateOptions(opts, args)

    cmd.Execute(opts, args)


def _status_lines(output: str) -> List[str]:
    """Normalize path separators and split command output into lines."""
    return output.replace(os.sep, "/").splitlines()


def _assert_project_header(line: str, project_path: str, branch: str) -> None:
    """Assert a status project header line for a project and branch."""
    expected = f"project {(project_path + '/ '):<40}branch {branch}"
    assert line == expected


def _assert_project_header_with_ahead_behind(
    line: str,
    project_path: str,
    branch: str,
    ahead_behind: str,
) -> None:
    """Assert a status project header line includes ahead/behind info."""
    suffix = f"branch {branch}{ahead_behind}"
    expected = f"project {(project_path + '/ '):<40}{suffix}"
    assert line == expected


def _assert_orphan_block(lines: List[str], expected: List[str]) -> None:
    """Assert orphan block header and entries, independent of entry ordering."""
    assert lines
    assert lines[0] == ("Objects not within a project (orphans)")
    orphan_lines = lines[1:]
    assert len(orphan_lines) == len(expected)
    assert sorted(orphan_lines) == sorted(expected)


def test_orphans_basic(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify -o output includes project header and orphan block."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))

    (topdir / "src" / "orphan_dir").mkdir(parents=True)
    (topdir / "orphan.txt").write_text("data")

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, ["-o"])

    lines = _status_lines(stdout.getvalue())
    _assert_project_header(lines[0], project_path, "main")
    _assert_orphan_block(
        lines[1:],
        [
            " --\torphan.txt",
            " --\tsrc/orphan_dir/",
        ],
    )


def test_empty_status_without_orphans(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify clean status without -o prints only the project header line."""
    _, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header(lines[0], project_path, "main")


def test_status_without_orphans(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify modified tracked file appears in status output without -o."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))

    (topdir / project_path / "README").write_text("updated")

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 2
    _assert_project_header(lines[0], project_path, "main")
    assert lines[1] == " -m\tREADME"


def test_status_with_orphans_and_modified_file(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify modified-file status plus orphan block."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))

    (topdir / project_path / "README").write_text("updated")
    (topdir / "src" / "orphan_dir").mkdir(parents=True)
    (topdir / "orphan.txt").write_text("data")

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, ["-o"])

    lines = _status_lines(stdout.getvalue())
    _assert_project_header(lines[0], project_path, "main")
    assert lines[1] == " -m\tREADME"
    _assert_orphan_block(
        lines[2:],
        [
            " --\torphan.txt",
            " --\tsrc/orphan_dir/",
        ],
    )


def test_empty_status_after_start_shows_started_branch(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify status shows the started branch name when the tree is clean."""
    topdir, manifest = repo_client_checkout

    project_path = next(iter(manifest.paths.keys()))
    project_worktree = topdir / project_path
    started_branch = "topic/test-status-branch"
    subprocess.check_call(
        ["git", "checkout", "-q", "-b", started_branch], cwd=project_worktree
    )

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header(lines[0], project_path, started_branch)


def _setup_remote_tracking_branch(
    manifest: manifest_xml.XmlManifest,
    branch_name: str,
) -> None:
    """Create a branch tracking a remote ref, like ``repo start``.

    In a real repo checkout, ``repo start`` creates a branch that
    tracks a remote tracking ref (e.g. refs/remotes/origin/main).
    This sets up the same config in both the worktree and the
    project gitdir (where repo reads its config from).
    """
    proj = list(manifest.paths.values())[0]
    worktree = Path(proj.worktree)
    proj_gitdir = Path(proj.gitdir)

    # Create the remote tracking ref in the worktree.
    subprocess.check_call(
        ["git", "update-ref", "refs/remotes/origin/main", "main"],
        cwd=worktree,
    )
    # Create the new branch from main in the worktree.
    subprocess.check_call(
        ["git", "checkout", "-q", "-b", branch_name, "main"],
        cwd=worktree,
    )
    # Write remote and branch config into the *project gitdir* config,
    # which is where repo's Project.config reads from.
    cfg = str(proj_gitdir / "config")
    subprocess.check_call(
        [
            "git",
            "config",
            "-f",
            cfg,
            "remote.origin.url",
            "http://localhost/fake",
        ],
    )
    subprocess.check_call(
        [
            "git",
            "config",
            "-f",
            cfg,
            "remote.origin.fetch",
            "+refs/heads/*:refs/remotes/origin/*",
        ],
    )
    subprocess.check_call(
        ["git", "config", "-f", cfg, f"branch.{branch_name}.remote", "origin"],
    )
    subprocess.check_call(
        [
            "git",
            "config",
            "-f",
            cfg,
            f"branch.{branch_name}.merge",
            "refs/heads/main",
        ],
    )


def test_status_branch_ahead_of_upstream(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify status shows [ahead N] for local commits."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))
    project_worktree = topdir / project_path

    _setup_remote_tracking_branch(manifest, "feature")
    subprocess.check_call(
        ["git", "commit", "-q", "--allow-empty", "-m", "c1"],
        cwd=project_worktree,
    )
    subprocess.check_call(
        ["git", "commit", "-q", "--allow-empty", "-m", "c2"],
        cwd=project_worktree,
    )

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header_with_ahead_behind(
        lines[0], project_path, "feature", " [ahead 2]"
    )


def test_status_branch_behind_upstream(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify status shows [behind N] when upstream is ahead."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))
    project_worktree = topdir / project_path

    _setup_remote_tracking_branch(manifest, "feature")
    # Advance the remote tracking ref past the feature branch.
    subprocess.check_call(
        ["git", "checkout", "-q", "main"], cwd=project_worktree
    )
    subprocess.check_call(
        ["git", "commit", "-q", "--allow-empty", "-m", "upstream"],
        cwd=project_worktree,
    )
    subprocess.check_call(
        ["git", "update-ref", "refs/remotes/origin/main", "main"],
        cwd=project_worktree,
    )
    subprocess.check_call(
        ["git", "checkout", "-q", "feature"],
        cwd=project_worktree,
    )

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header_with_ahead_behind(
        lines[0], project_path, "feature", " [behind 1]"
    )


def test_status_branch_ahead_and_behind(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify [ahead N, behind M] when branch has diverged."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))
    project_worktree = topdir / project_path

    _setup_remote_tracking_branch(manifest, "feature")
    # Add a local commit on feature.
    subprocess.check_call(
        ["git", "commit", "-q", "--allow-empty", "-m", "local"],
        cwd=project_worktree,
    )
    # Advance the remote tracking ref independently.
    subprocess.check_call(
        ["git", "checkout", "-q", "main"], cwd=project_worktree
    )
    subprocess.check_call(
        ["git", "commit", "-q", "--allow-empty", "-m", "upstream"],
        cwd=project_worktree,
    )
    subprocess.check_call(
        ["git", "update-ref", "refs/remotes/origin/main", "main"],
        cwd=project_worktree,
    )
    subprocess.check_call(
        ["git", "checkout", "-q", "feature"],
        cwd=project_worktree,
    )

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header_with_ahead_behind(
        lines[0],
        project_path,
        "feature",
        " [ahead 1, behind 1]",
    )


def test_status_branch_no_tracking_no_ahead_behind(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify no ahead/behind when branch has no upstream."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))
    project_worktree = topdir / project_path

    subprocess.check_call(
        [
            "git",
            "checkout",
            "-q",
            "-b",
            "orphan-branch",
            "--no-track",
            "main",
        ],
        cwd=project_worktree,
    )
    subprocess.check_call(
        ["git", "commit", "-q", "--allow-empty", "-m", "c1"],
        cwd=project_worktree,
    )

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header(lines[0], project_path, "orphan-branch")


def test_status_branch_synced_no_ahead_behind(
    repo_client_checkout: Tuple[Path, manifest_xml.XmlManifest],
) -> None:
    """Verify no ahead/behind when branch is fully synced."""
    topdir, manifest = repo_client_checkout
    project_path = next(iter(manifest.paths.keys()))

    _setup_remote_tracking_branch(manifest, "synced")

    with contextlib.redirect_stdout(io.StringIO()) as stdout:
        _run_status(manifest, [])

    lines = _status_lines(stdout.getvalue())
    assert len(lines) == 1
    _assert_project_header(lines[0], project_path, "synced")