"""Tests for LangGraph state module. Written FIRST (TDD RED phase).""" import json from release_agent.state import ReleaseState, add_errors, add_messages # --------------------------------------------------------------------------- # ReleaseState tests # --------------------------------------------------------------------------- class TestReleaseState: """Tests for ReleaseState TypedDict.""" def test_empty_state_is_valid(self) -> None: # total=False means all fields are optional state: ReleaseState = {} assert state == {} def test_partial_state_with_repo(self) -> None: state: ReleaseState = {"repo_name": "my-repo"} assert state["repo_name"] == "my-repo" def test_partial_state_with_messages(self) -> None: state: ReleaseState = {"messages": ["Hello"]} assert state["messages"] == ["Hello"] def test_partial_state_with_errors(self) -> None: state: ReleaseState = {"errors": ["Something went wrong"]} assert state["errors"] == ["Something went wrong"] def test_state_with_pr_id(self) -> None: state: ReleaseState = {"pr_id": "PR-42"} assert state["pr_id"] == "PR-42" def test_state_with_ticket_id(self) -> None: state: ReleaseState = {"ticket_id": "ALLPOST-100"} assert state["ticket_id"] == "ALLPOST-100" def test_state_with_version(self) -> None: state: ReleaseState = {"version": "v1.0.1"} assert state["version"] == "v1.0.1" # Phase 3 new fields def test_state_with_webhook_payload(self) -> None: state: ReleaseState = {"webhook_payload": {"event_type": "git.pullrequest.merged"}} assert state["webhook_payload"]["event_type"] == "git.pullrequest.merged" def test_state_with_pr_info(self) -> None: state: ReleaseState = {"pr_info": {"pr_id": "42", "repo_name": "my-repo"}} assert state["pr_info"]["repo_name"] == "my-repo" def test_state_with_pr_diff(self) -> None: state: ReleaseState = {"pr_diff": "edit: src/main.py"} assert state["pr_diff"] == "edit: src/main.py" def test_state_with_last_merge_source_commit(self) -> None: state: ReleaseState = {"last_merge_source_commit": "abc123"} assert state["last_merge_source_commit"] == "abc123" def test_state_with_ticket_summary(self) -> None: state: ReleaseState = {"ticket_summary": "Fix login bug"} assert state["ticket_summary"] == "Fix login bug" def test_state_with_has_ticket(self) -> None: state: ReleaseState = {"has_ticket": True} assert state["has_ticket"] is True def test_state_with_review_result(self) -> None: state: ReleaseState = {"review_result": {"verdict": "approve", "summary": "LGTM"}} assert state["review_result"]["verdict"] == "approve" def test_state_with_review_approved(self) -> None: state: ReleaseState = {"review_approved": True} assert state["review_approved"] is True def test_state_with_staging(self) -> None: state: ReleaseState = {"staging": {"version": "v1.0.0", "tickets": []}} assert state["staging"]["version"] == "v1.0.0" def test_state_with_pr_already_merged(self) -> None: state: ReleaseState = {"pr_already_merged": False} assert state["pr_already_merged"] is False def test_state_with_release_pr_id(self) -> None: state: ReleaseState = {"release_pr_id": "123"} assert state["release_pr_id"] == "123" def test_state_with_release_pr_commit(self) -> None: state: ReleaseState = {"release_pr_commit": "deadbeef"} assert state["release_pr_commit"] == "deadbeef" def test_state_with_pipelines(self) -> None: state: ReleaseState = {"pipelines": [{"id": 1, "name": "build"}]} assert len(state["pipelines"]) == 1 def test_state_with_triggered_builds(self) -> None: state: ReleaseState = {"triggered_builds": [{"id": 99}]} assert state["triggered_builds"][0]["id"] == 99 def test_state_with_pending_approvals(self) -> None: state: ReleaseState = {"pending_approvals": [{"approval_id": "aaa"}]} assert state["pending_approvals"][0]["approval_id"] == "aaa" def test_state_with_continue_to_release(self) -> None: state: ReleaseState = {"continue_to_release": True} assert state["continue_to_release"] is True # Phase 5: CI/CD and approval fields def test_state_with_ci_build_id(self) -> None: state: ReleaseState = {"ci_build_id": 12345} assert state["ci_build_id"] == 12345 def test_state_with_ci_build_status(self) -> None: state: ReleaseState = {"ci_build_status": "inProgress"} assert state["ci_build_status"] == "inProgress" def test_state_with_ci_build_result(self) -> None: state: ReleaseState = {"ci_build_result": "succeeded"} assert state["ci_build_result"] == "succeeded" def test_state_with_ci_build_url(self) -> None: state: ReleaseState = {"ci_build_url": "https://dev.azure.com/org/proj/_build/results?buildId=99"} assert "buildId=99" in state["ci_build_url"] def test_state_with_release_definition_id(self) -> None: state: ReleaseState = {"release_definition_id": 7} assert state["release_definition_id"] == 7 def test_state_with_release_id(self) -> None: state: ReleaseState = {"release_id": 456} assert state["release_id"] == 456 def test_state_with_current_stage(self) -> None: state: ReleaseState = {"current_stage": "sandbox_pending"} assert state["current_stage"] == "sandbox_pending" def test_state_with_approval_message_ts(self) -> None: state: ReleaseState = {"approval_message_ts": "1234567890.123456"} assert state["approval_message_ts"] == "1234567890.123456" def test_state_with_slack_message_ts(self) -> None: state: ReleaseState = {"slack_message_ts": "9876543210.000001"} assert state["slack_message_ts"] == "9876543210.000001" def test_state_json_serializable_empty(self) -> None: state: ReleaseState = {} serialized = json.dumps(state) assert json.loads(serialized) == {} def test_state_json_serializable_with_strings(self) -> None: state: ReleaseState = { "repo_name": "my-repo", "pr_id": "PR-1", "ticket_id": "ALLPOST-1", "version": "v1.0.0", } serialized = json.dumps(state) loaded = json.loads(serialized) assert loaded["repo_name"] == "my-repo" assert loaded["pr_id"] == "PR-1" def test_state_json_serializable_with_lists(self) -> None: state: ReleaseState = { "messages": ["msg1", "msg2"], "errors": ["err1"], } serialized = json.dumps(state) loaded = json.loads(serialized) assert loaded["messages"] == ["msg1", "msg2"] assert loaded["errors"] == ["err1"] # --------------------------------------------------------------------------- # Reducer tests # --------------------------------------------------------------------------- class TestAddMessages: """Tests for add_messages reducer.""" def test_accumulates_to_empty(self) -> None: result = add_messages([], ["Hello"]) assert result == ["Hello"] def test_accumulates_to_existing(self) -> None: result = add_messages(["Hello"], ["World"]) assert result == ["Hello", "World"] def test_accumulates_multiple(self) -> None: result = add_messages(["A", "B"], ["C", "D"]) assert result == ["A", "B", "C", "D"] def test_existing_unchanged(self) -> None: existing = ["Hello"] add_messages(existing, ["World"]) # Original should not be mutated assert existing == ["Hello"] def test_empty_new_messages(self) -> None: result = add_messages(["Hello"], []) assert result == ["Hello"] def test_both_empty(self) -> None: result = add_messages([], []) assert result == [] def test_returns_new_list(self) -> None: existing = ["Hello"] new_msgs = ["World"] result = add_messages(existing, new_msgs) assert result is not existing assert result is not new_msgs class TestAddErrors: """Tests for add_errors reducer.""" def test_accumulates_to_empty(self) -> None: result = add_errors([], ["Error occurred"]) assert result == ["Error occurred"] def test_accumulates_to_existing(self) -> None: result = add_errors(["First error"], ["Second error"]) assert result == ["First error", "Second error"] def test_existing_unchanged(self) -> None: existing = ["First error"] add_errors(existing, ["Second error"]) assert existing == ["First error"] def test_empty_new_errors(self) -> None: result = add_errors(["Existing"], []) assert result == ["Existing"] def test_both_empty(self) -> None: result = add_errors([], []) assert result == [] def test_returns_new_list(self) -> None: existing = ["Error"] result = add_errors(existing, ["New error"]) assert result is not existing