208 lines
6.9 KiB
Python
208 lines
6.9 KiB
Python
"""
|
|
test_collaboration_service.py — Unit tests for three-way diff, fork, merge request lifecycle.
|
|
"""
|
|
import asyncio
|
|
import copy
|
|
import sys
|
|
import os
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
|
|
from oracle.collaboration_service import CollaborationService, three_way_diff
|
|
|
|
|
|
def run(coro):
|
|
return asyncio.get_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def _comp(cid, title="Test", order=100, content="default"):
|
|
return {
|
|
"componentId": cid,
|
|
"type": "barChart",
|
|
"title": title,
|
|
"dataSourceDescriptor": {"dataset": "test_ds", "queryTemplate": content},
|
|
"accessControls": {"allowedRoles": ["sales_director"], "visibilityScope": "private"},
|
|
"layout": {"orderIndex": order, "sectionId": "sec_test", "widthMode": "full"},
|
|
}
|
|
|
|
|
|
# ── Three-way diff tests ──────────────────────────────────────────────────────
|
|
|
|
def test_safe_append_in_source():
|
|
base = [_comp("cmp_a")]
|
|
source = [_comp("cmp_a"), _comp("cmp_b")] # cmp_b added in source
|
|
target = [_comp("cmp_a")]
|
|
|
|
merged, conflicts = three_way_diff(base, source, target)
|
|
assert any(c["conflictClass"] == "safe_append" and c["componentId"] == "cmp_b" for c in conflicts)
|
|
assert any(c["componentId"] == "cmp_b" for c in merged)
|
|
|
|
|
|
def test_no_conflict_when_identical():
|
|
base = [_comp("cmp_a")]
|
|
source = [_comp("cmp_a")]
|
|
target = [_comp("cmp_a")]
|
|
|
|
merged, conflicts = three_way_diff(base, source, target)
|
|
assert len(merged) == 1
|
|
assert all(c["conflictClass"] not in ("component_content_conflict", "query_descriptor_conflict") for c in conflicts)
|
|
|
|
|
|
def test_component_content_conflict():
|
|
base = [_comp("cmp_a", content="SELECT 1")]
|
|
source = [_comp("cmp_a", content="SELECT 2")]
|
|
target = [_comp("cmp_a", content="SELECT 3")]
|
|
|
|
merged, conflicts = three_way_diff(base, source, target)
|
|
# Expect query_descriptor_conflict or component_content_conflict
|
|
conflict_classes = {c["conflictClass"] for c in conflicts}
|
|
assert conflict_classes & {"component_content_conflict", "query_descriptor_conflict"}
|
|
|
|
|
|
def test_delete_edit_conflict_source_deletes():
|
|
base = [_comp("cmp_a"), _comp("cmp_b")]
|
|
source = [_comp("cmp_a")] # cmp_b deleted in source
|
|
target = [_comp("cmp_a"), _comp("cmp_b", title="Edited in target")] # cmp_b edited in target
|
|
|
|
merged, conflicts = three_way_diff(base, source, target)
|
|
assert any(c["conflictClass"] == "delete_edit_conflict" and c["componentId"] == "cmp_b" for c in conflicts)
|
|
# Default: keep target (edited version)
|
|
assert any(c["componentId"] == "cmp_b" for c in merged)
|
|
|
|
|
|
def test_deleted_in_both_is_removed():
|
|
base = [_comp("cmp_a"), _comp("cmp_b")]
|
|
source = [_comp("cmp_a")]
|
|
target = [_comp("cmp_a")]
|
|
|
|
merged, conflicts = three_way_diff(base, source, target)
|
|
assert not any(c["componentId"] == "cmp_b" for c in merged)
|
|
|
|
|
|
def test_orderindex_normalization():
|
|
base = []
|
|
source = [_comp("c1", order=100), _comp("c2", order=200)]
|
|
target = [_comp("c1", order=100), _comp("c2", order=200)]
|
|
|
|
merged, _ = three_way_diff(base, source, target)
|
|
orders = [c["layout"]["orderIndex"] for c in merged]
|
|
# Orders should be normalized (multiples of 100, sequential)
|
|
assert orders == sorted(orders)
|
|
assert all(o % 100 == 0 for o in orders)
|
|
|
|
|
|
# ── CollaborationService tests ────────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def collab():
|
|
from oracle import collaboration_service as _mod
|
|
_mod._DEMO_FORKS.clear()
|
|
_mod._DEMO_MRS.clear()
|
|
return CollaborationService()
|
|
|
|
|
|
def test_create_fork(collab):
|
|
source_page = {
|
|
"pageId": "page_src",
|
|
"branchId": "branch_main",
|
|
"headRevision": 5,
|
|
"components": [],
|
|
}
|
|
fork = run(collab.create_fork(
|
|
source_page=source_page,
|
|
recipient_user_id="user_recipient",
|
|
created_by="user_src",
|
|
))
|
|
assert fork["sourcePageId"] == "page_src"
|
|
assert fork["sourceRevision"] == 5
|
|
assert fork["status"] == "active"
|
|
assert fork["recipientUserId"] == "user_recipient"
|
|
|
|
|
|
def test_merge_request_lifecycle(collab):
|
|
mr = run(collab.open_merge_request(
|
|
tenant_id="tenant_test",
|
|
source_page_id="page_fork",
|
|
source_branch_id="branch_fork",
|
|
source_head_revision=2,
|
|
target_page_id="page_main",
|
|
target_branch_id="branch_main",
|
|
target_base_revision=5,
|
|
title="Test MR",
|
|
description="My changes",
|
|
created_by="user_a",
|
|
source_components=[_comp("cmp_a"), _comp("cmp_new")],
|
|
target_components=[_comp("cmp_a")],
|
|
base_components=[_comp("cmp_a")],
|
|
))
|
|
assert mr["status"] == "open"
|
|
assert "mergeRequestId" in mr
|
|
|
|
# Approve it
|
|
reviewed = run(collab.review_merge_request(
|
|
mr_id=mr["mergeRequestId"],
|
|
decision="approve",
|
|
reviewer_id="user_reviewer",
|
|
))
|
|
assert reviewed["status"] == "merged"
|
|
|
|
|
|
def test_merge_request_reject(collab):
|
|
mr = run(collab.open_merge_request(
|
|
tenant_id="tenant_test",
|
|
source_page_id="page_fork",
|
|
source_branch_id="branch_fork",
|
|
source_head_revision=1,
|
|
target_page_id="page_main",
|
|
target_branch_id="branch_main",
|
|
target_base_revision=1,
|
|
title="Rejected MR",
|
|
created_by="user_a",
|
|
source_components=[],
|
|
target_components=[],
|
|
base_components=[],
|
|
))
|
|
reviewed = run(collab.review_merge_request(
|
|
mr_id=mr["mergeRequestId"],
|
|
decision="reject",
|
|
reviewer_id="user_reviewer",
|
|
))
|
|
assert reviewed["status"] == "closed"
|
|
|
|
|
|
def test_list_merge_requests_filters_by_target(collab):
|
|
for i in range(3):
|
|
run(collab.open_merge_request(
|
|
tenant_id="tenant_t",
|
|
source_page_id=f"page_fork_{i}",
|
|
source_branch_id=f"branch_{i}",
|
|
source_head_revision=1,
|
|
target_page_id="page_target",
|
|
target_branch_id="branch_main",
|
|
target_base_revision=1,
|
|
title=f"MR {i}",
|
|
created_by="user_a",
|
|
source_components=[],
|
|
target_components=[],
|
|
base_components=[],
|
|
))
|
|
# Different target
|
|
run(collab.open_merge_request(
|
|
tenant_id="tenant_t",
|
|
source_page_id="page_other",
|
|
source_branch_id="branch_other",
|
|
source_head_revision=1,
|
|
target_page_id="page_other_target",
|
|
target_branch_id="branch_main",
|
|
target_base_revision=1,
|
|
title="Different target MR",
|
|
created_by="user_b",
|
|
source_components=[],
|
|
target_components=[],
|
|
base_components=[],
|
|
))
|
|
mrs = run(collab.list_merge_requests("page_target"))
|
|
assert len(mrs) == 3
|
|
assert all(mr["targetPageId"] == "page_target" for mr in mrs)
|