diff --git a/pygittools/auth_test.py b/pygittools/auth_test.py new file mode 100644 index 0000000..57a5803 --- /dev/null +++ b/pygittools/auth_test.py @@ -0,0 +1,35 @@ +import time +from datetime import timedelta + +from pygittools.auth import RootAuthProvider + + +def test_rootauthprovider_session() -> None: + provider = RootAuthProvider(admin_user=b"root", admin_password=b"secret") + + session = provider.create_session(user=b"root", password=b"secret") + + assert session is not None + assert provider.validate_session(session) is True + + +def test_rootauthprovider_badpass() -> None: + provider = RootAuthProvider(admin_user=b"root", admin_password=b"secret") + + session = provider.create_session(user=b"root", password=b"wrong") + + assert session is None + + +def test_rootauthprovider_timeout() -> None: + provider = RootAuthProvider( + admin_user=b"root", + admin_password=b"secret", + session_timeout=timedelta(milliseconds=1), + ) + + session = provider.create_session(user=b"root", password=b"secret") + + assert session is not None + time.sleep(0.005) + assert provider.validate_session(session) is False diff --git a/pygittools/tasks_test.py b/pygittools/tasks_test.py new file mode 100644 index 0000000..57e63c5 --- /dev/null +++ b/pygittools/tasks_test.py @@ -0,0 +1,203 @@ +import subprocess +from pathlib import Path + +import pytest +from pygit2 import Oid, Repository, init_repository + +from pygittools.tasks import ( + BOARD_REF_PREFIX, + EMPTY_TREE_OID_HEX, + TASK_REF_PREFIX, + Board, + Comment, + Task, + get_board, + get_comment, + get_task, + get_task_by_oid, +) + +TAGGER: str = "alice " +BOARD_MAIN: str = f"{BOARD_REF_PREFIX}main" + + +@pytest.fixture +def repo(tmp_path: Path) -> Repository: + repo_path = tmp_path / "repo" + repo_path.mkdir() + return init_repository(str(repo_path), bare=False) + + +def _empty_tree() -> Oid: + return Oid(hex=EMPTY_TREE_OID_HEX) + + +def _task_ref(slug: str) -> str: + return f"{TASK_REF_PREFIX}{slug}" + + +def _run_git_gc(repo: Repository) -> None: + subprocess.run( + ["git", "gc", "--prune=now"], + cwd=repo.workdir, + capture_output=True, + text=True, + check=True, + ) + + +def test_board_roundtrip(repo: Repository) -> None: + board = Board(_empty_tree(), BOARD_MAIN, TAGGER, description="Main board") + board.write(repo) + + loaded = get_board(repo, BOARD_MAIN) + assert loaded is not None + assert loaded.name == BOARD_MAIN + assert loaded.description == "Main board" + assert loaded.tasks == [] + + +def test_task_roundtrip_by_ref_and_oid(repo: Repository) -> None: + task = Task( + _empty_tree(), + _task_ref("alpha"), + TAGGER, + title="Alpha", + description="First task", + status=Task.Status.TODO, + priority=Task.Priority.HIGH, + assignee="bob", + ) + oid = task.write(repo) + + by_ref = get_task(repo, _task_ref("alpha")) + assert by_ref is not None + assert by_ref.title == "Alpha" + assert by_ref.description == "First task" + assert by_ref.status == Task.Status.TODO + assert by_ref.priority == Task.Priority.HIGH + assert by_ref.assignee == "bob" + + by_oid = get_task_by_oid(repo, oid) + assert by_oid is not None + assert by_oid.title == "Alpha" + assert by_oid.status == Task.Status.TODO + + +def test_comment_roundtrip(repo: Repository) -> None: + task = Task(_empty_tree(), _task_ref("with-comment"), TAGGER, title="WithComment") + task_oid = task.write(repo) + + comment = Comment(task_oid, "bob ", content="Hello there") + c_oid = comment.write(repo) + + loaded = get_comment(repo, c_oid) + assert loaded is not None + assert loaded.content == "Hello there" + assert loaded.target == task_oid + + +def test_full_board_flow(repo: Repository) -> None: + board = Board(_empty_tree(), BOARD_MAIN, TAGGER, description="Main") + + task_a = Task( + _empty_tree(), + _task_ref("a"), + TAGGER, + title="Task A", + description="Desc A", + status=Task.Status.TODO, + priority=Task.Priority.HIGH, + ) + task_a_oid = task_a.write(repo) + + task_b = Task( + _empty_tree(), + _task_ref("b"), + TAGGER, + title="Task B", + description="Desc B", + status=Task.Status.IN_PROGRESS, + priority=Task.Priority.MEDIUM, + ) + task_b_oid = task_b.write(repo) + + comment_a1 = Comment(task_a_oid, "bob ", content="A1") + a1_oid = comment_a1.write(repo) + comment_a2 = Comment(task_a_oid, "carol ", content="A2") + a2_oid = comment_a2.write(repo) + + task_a.comments = [str(a1_oid), str(a2_oid)] + task_a.update_message() + task_a_oid = task_a.write(repo) + + comment_b1 = Comment(task_b_oid, "dave ", content="B1") + b1_oid = comment_b1.write(repo) + + task_b.comments = [str(b1_oid)] + task_b.update_message() + task_b_oid = task_b.write(repo) + + board.tasks = [str(task_a_oid), str(task_b_oid)] + board.update_message() + board.write(repo) + + loaded_board = get_board(repo, BOARD_MAIN) + assert loaded_board is not None + assert len(loaded_board.tasks) == 2 + + loaded_a = get_task_by_oid(repo, Oid(hex=loaded_board.tasks[0])) + assert loaded_a is not None + assert loaded_a.title == "Task A" + assert loaded_a.priority == Task.Priority.HIGH + assert [get_comment(repo, Oid(hex=c)).content for c in loaded_a.comments] == ["A1", "A2"] + + loaded_b = get_task_by_oid(repo, Oid(hex=loaded_board.tasks[1])) + assert loaded_b is not None + assert loaded_b.title == "Task B" + assert loaded_b.status == Task.Status.IN_PROGRESS + assert [get_comment(repo, Oid(hex=c)).content for c in loaded_b.comments] == ["B1"] + + +def test_full_board_flow_survives_git_gc(repo: Repository, tmp_path: Path) -> None: + board = Board(_empty_tree(), BOARD_MAIN, TAGGER, description="Main") + + task = Task( + _empty_tree(), + _task_ref("gc-test"), + TAGGER, + title="GC Task", + description="persists across gc", + status=Task.Status.TODO, + priority=Task.Priority.CRITICAL, + ) + task_oid = task.write(repo) + + c1_oid = Comment(task_oid, "bob ", content="first").write(repo) + c2_oid = Comment(task_oid, "carol ", content="second").write(repo) + + task.comments = [str(c1_oid), str(c2_oid)] + task.update_message() + task_oid = task.write(repo) + + board.tasks = [str(task_oid)] + board.update_message() + board.write(repo) + + _run_git_gc(repo) + + # Re-open repo to avoid any in-memory odb caches from the pre-gc instance. + reopened = Repository(repo.path) + + loaded_board = get_board(reopened, BOARD_MAIN) + assert loaded_board is not None + assert len(loaded_board.tasks) == 1 + + loaded_task = get_task_by_oid(reopened, Oid(hex=loaded_board.tasks[0])) + assert loaded_task is not None + assert loaded_task.title == "GC Task" + assert loaded_task.priority == Task.Priority.CRITICAL + assert len(loaded_task.comments) == 2 + + contents = [get_comment(reopened, Oid(hex=c)).content for c in loaded_task.comments] + assert contents == ["first", "second"] diff --git a/pygittools/test_tasks.py b/pygittools/test_tasks.py deleted file mode 100644 index 57e63c5..0000000 --- a/pygittools/test_tasks.py +++ /dev/null @@ -1,203 +0,0 @@ -import subprocess -from pathlib import Path - -import pytest -from pygit2 import Oid, Repository, init_repository - -from pygittools.tasks import ( - BOARD_REF_PREFIX, - EMPTY_TREE_OID_HEX, - TASK_REF_PREFIX, - Board, - Comment, - Task, - get_board, - get_comment, - get_task, - get_task_by_oid, -) - -TAGGER: str = "alice " -BOARD_MAIN: str = f"{BOARD_REF_PREFIX}main" - - -@pytest.fixture -def repo(tmp_path: Path) -> Repository: - repo_path = tmp_path / "repo" - repo_path.mkdir() - return init_repository(str(repo_path), bare=False) - - -def _empty_tree() -> Oid: - return Oid(hex=EMPTY_TREE_OID_HEX) - - -def _task_ref(slug: str) -> str: - return f"{TASK_REF_PREFIX}{slug}" - - -def _run_git_gc(repo: Repository) -> None: - subprocess.run( - ["git", "gc", "--prune=now"], - cwd=repo.workdir, - capture_output=True, - text=True, - check=True, - ) - - -def test_board_roundtrip(repo: Repository) -> None: - board = Board(_empty_tree(), BOARD_MAIN, TAGGER, description="Main board") - board.write(repo) - - loaded = get_board(repo, BOARD_MAIN) - assert loaded is not None - assert loaded.name == BOARD_MAIN - assert loaded.description == "Main board" - assert loaded.tasks == [] - - -def test_task_roundtrip_by_ref_and_oid(repo: Repository) -> None: - task = Task( - _empty_tree(), - _task_ref("alpha"), - TAGGER, - title="Alpha", - description="First task", - status=Task.Status.TODO, - priority=Task.Priority.HIGH, - assignee="bob", - ) - oid = task.write(repo) - - by_ref = get_task(repo, _task_ref("alpha")) - assert by_ref is not None - assert by_ref.title == "Alpha" - assert by_ref.description == "First task" - assert by_ref.status == Task.Status.TODO - assert by_ref.priority == Task.Priority.HIGH - assert by_ref.assignee == "bob" - - by_oid = get_task_by_oid(repo, oid) - assert by_oid is not None - assert by_oid.title == "Alpha" - assert by_oid.status == Task.Status.TODO - - -def test_comment_roundtrip(repo: Repository) -> None: - task = Task(_empty_tree(), _task_ref("with-comment"), TAGGER, title="WithComment") - task_oid = task.write(repo) - - comment = Comment(task_oid, "bob ", content="Hello there") - c_oid = comment.write(repo) - - loaded = get_comment(repo, c_oid) - assert loaded is not None - assert loaded.content == "Hello there" - assert loaded.target == task_oid - - -def test_full_board_flow(repo: Repository) -> None: - board = Board(_empty_tree(), BOARD_MAIN, TAGGER, description="Main") - - task_a = Task( - _empty_tree(), - _task_ref("a"), - TAGGER, - title="Task A", - description="Desc A", - status=Task.Status.TODO, - priority=Task.Priority.HIGH, - ) - task_a_oid = task_a.write(repo) - - task_b = Task( - _empty_tree(), - _task_ref("b"), - TAGGER, - title="Task B", - description="Desc B", - status=Task.Status.IN_PROGRESS, - priority=Task.Priority.MEDIUM, - ) - task_b_oid = task_b.write(repo) - - comment_a1 = Comment(task_a_oid, "bob ", content="A1") - a1_oid = comment_a1.write(repo) - comment_a2 = Comment(task_a_oid, "carol ", content="A2") - a2_oid = comment_a2.write(repo) - - task_a.comments = [str(a1_oid), str(a2_oid)] - task_a.update_message() - task_a_oid = task_a.write(repo) - - comment_b1 = Comment(task_b_oid, "dave ", content="B1") - b1_oid = comment_b1.write(repo) - - task_b.comments = [str(b1_oid)] - task_b.update_message() - task_b_oid = task_b.write(repo) - - board.tasks = [str(task_a_oid), str(task_b_oid)] - board.update_message() - board.write(repo) - - loaded_board = get_board(repo, BOARD_MAIN) - assert loaded_board is not None - assert len(loaded_board.tasks) == 2 - - loaded_a = get_task_by_oid(repo, Oid(hex=loaded_board.tasks[0])) - assert loaded_a is not None - assert loaded_a.title == "Task A" - assert loaded_a.priority == Task.Priority.HIGH - assert [get_comment(repo, Oid(hex=c)).content for c in loaded_a.comments] == ["A1", "A2"] - - loaded_b = get_task_by_oid(repo, Oid(hex=loaded_board.tasks[1])) - assert loaded_b is not None - assert loaded_b.title == "Task B" - assert loaded_b.status == Task.Status.IN_PROGRESS - assert [get_comment(repo, Oid(hex=c)).content for c in loaded_b.comments] == ["B1"] - - -def test_full_board_flow_survives_git_gc(repo: Repository, tmp_path: Path) -> None: - board = Board(_empty_tree(), BOARD_MAIN, TAGGER, description="Main") - - task = Task( - _empty_tree(), - _task_ref("gc-test"), - TAGGER, - title="GC Task", - description="persists across gc", - status=Task.Status.TODO, - priority=Task.Priority.CRITICAL, - ) - task_oid = task.write(repo) - - c1_oid = Comment(task_oid, "bob ", content="first").write(repo) - c2_oid = Comment(task_oid, "carol ", content="second").write(repo) - - task.comments = [str(c1_oid), str(c2_oid)] - task.update_message() - task_oid = task.write(repo) - - board.tasks = [str(task_oid)] - board.update_message() - board.write(repo) - - _run_git_gc(repo) - - # Re-open repo to avoid any in-memory odb caches from the pre-gc instance. - reopened = Repository(repo.path) - - loaded_board = get_board(reopened, BOARD_MAIN) - assert loaded_board is not None - assert len(loaded_board.tasks) == 1 - - loaded_task = get_task_by_oid(reopened, Oid(hex=loaded_board.tasks[0])) - assert loaded_task is not None - assert loaded_task.title == "GC Task" - assert loaded_task.priority == Task.Priority.CRITICAL - assert len(loaded_task.comments) == 2 - - contents = [get_comment(reopened, Oid(hex=c)).content for c in loaded_task.comments] - assert contents == ["first", "second"]