[worktree-unit-tests-reenable] worktree-tests: Mocked

tasks/libs/common/worktree.py

@@ -63,7 +63,7 @@ def remove_env(ctx):
 def is_worktree():
     """Will return True if the current environment is a worktree environment."""
 
-    return Path.cwd() == WORKTREE_DIRECTORY
+    return Path.cwd().resolve() == WORKTREE_DIRECTORY.resolve()
 
 
 def enter_env(ctx, branch: str | None, skip_checkout=False):

tasks/unit_tests/libs/common/worktree_tests.py

@@ -1,104 +1,147 @@
-# TODO(celian): Reintroduce these tests
+import os
+import re
+import shutil
+import unittest
+from pathlib import Path
+from unittest.mock import patch
 
-"""
-NOTE: These tests are disabled since they use git commands that are not mocked yet.
-This breaks unit tests ran with macos runners.
-"""
+from invoke.context import Context, MockContext, Result
 
-# import os
-# import unittest
+from tasks.libs.common.gomodules import Configuration, get_default_modules
+from tasks.libs.common.worktree import agent_context, init_env, is_worktree
 
-# from invoke import Context
+TEST_WORKTREE_DIR = Path('/tmp/datadog-agent-worktree')
 
-# from tasks.libs.common.git import get_default_branch
-# from tasks.libs.common.gomodules import get_default_modules
-# from tasks.libs.common.worktree import agent_context, init_env, is_worktree
 
+def get_ctx():
+    return Context()
 
-# def get_ctx():
-#     return Context()
 
+class WorktreeMockContext(MockContext):
+    """Mock for invoke context to simulate git commands (clone / switch).
 
-# class TestWorktree(unittest.TestCase):
-#     def setUp(self):
-#         # Pull only once
-#         init_env(get_ctx(), '6.53.x')
-#         os.environ['AGENT_WORKTREE_NO_PULL'] = '1'
+    Only `modules.yml` will be updated to simulate it.
+    """
 
-#     def test_context_is_worktree_true(self):
-#         with agent_context(get_ctx(), '6.53.x'):
-#             self.assertTrue(is_worktree())
+    def __init__(self):
+        super().__init__()
 
-#     def test_context_is_worktree_false(self):
-#         self.assertFalse(is_worktree())
+        self.branch = None
+        self.modules_main = get_default_modules()
+        # Exclude some modules
+        self.modules_653x = {name: mod for (name, mod) in get_default_modules().items() if not name.startswith('comp/')}
 
-#     def test_context_nested(self):
-#         with agent_context(get_ctx(), '6.53.x'):
-#             with agent_context(get_ctx(), '6.53.x'):
-#                 self.assertTrue(is_worktree())
-#             self.assertTrue(is_worktree())
+        assert self.modules_main.keys() != self.modules_653x.keys()
 
-#     def test_context_pwd(self):
-#         ctx = get_ctx()
+    def _checkout(self, branch):
+        assert branch in ('main', '6.53.x')
 
-#         with agent_context(ctx, None, skip_checkout=True):
-#             pwdnone = ctx.run('pwd').stdout
+        if self.branch == branch:
+            return
 
-#         with agent_context(ctx, '6.53.x'):
-#             pwd6 = ctx.run('pwd').stdout
+        self.branch = branch
+        modules = self.modules_main if branch == 'main' else self.modules_653x
 
-#         with agent_context(ctx, 'main'):
-#             pwdmain = ctx.run('pwd').stdout
+        Configuration(TEST_WORKTREE_DIR, modules, set()).to_file()
 
-#         self.assertEqual(pwd6, pwdnone)
-#         self.assertEqual(pwd6, pwdmain)
+    def _clone(self):
+        self.reset()
+        TEST_WORKTREE_DIR.mkdir(parents=True)
+        self._checkout('main')
 
-#     def test_context_modules(self):
-#         ctx = get_ctx()
+    def reset(self):
+        shutil.rmtree(TEST_WORKTREE_DIR, ignore_errors=True)
+        self.branch = None
 
-#         with agent_context(ctx, 'main'):
-#             modules7 = get_default_modules()
+    def run(self, command, *args, **kwargs):
+        if re.match(r'git.*rev-parse.*', command):
+            return Result(stdout=self.branch)
 
-#         with agent_context(ctx, '6.53.x'):
-#             modules6 = get_default_modules()
+        if re.match(r'git.*remote get-url.*', command):
+            return Result(stdout='[email protected]:datadog/datadog-agent.git')
 
-#         self.assertNotEqual(set(modules6.keys()), set(modules7.keys()))
+        if re.match(r'git.*clone.*', command):
+            self._clone()
 
-#     def test_context_branch(self):
-#         ctx = get_ctx()
+            return Result(stdout='cloned')
 
-#         with agent_context(ctx, 'main'):
-#             branch7 = get_default_branch()
+        if re.match(r'git.*checkout.*', command):
+            on_653x = '6.53.x' in command
 
-#         with agent_context(ctx, '6.53.x'):
-#             branch6 = get_default_branch()
+            self._checkout('6.53.x' if on_653x else 'main')
 
-#         self.assertNotEqual(branch6, branch7)
+            return Result(stdout='checked out')
 
-#     def test_context_no_checkout(self):
-#         ctx = get_ctx()
+        return super().run(command, *args, **kwargs)
 
-#         with agent_context(ctx, '6.53.x'):
-#             branch6 = get_default_branch()
 
-#         with agent_context(ctx, 'main'):
-#             branch7 = get_default_branch()
+class TestWorktree(unittest.TestCase):
+    def setUp(self):
+        # Pull only once
+        self.mock_ctx = WorktreeMockContext()
+        self.mock_ctx.reset()
+        self.patch_workdir = patch('tasks.libs.common.worktree.WORKTREE_DIRECTORY', TEST_WORKTREE_DIR)
+        self.patch_workdir.start()
+        self.patch_agentdir = patch('tasks.libs.common.gomodules.agent_working_directory', lambda: TEST_WORKTREE_DIR)
+        self.patch_agentdir.start()
 
-#         with agent_context(ctx, 'main', skip_checkout=True):
-#             branch_no_checkout = get_default_branch()
+        os.environ['AGENT_WORKTREE_NO_PULL'] = '1'
+        init_env(self.mock_ctx, '6.53.x')
 
-#         self.assertNotEqual(branch6, branch7)
-#         self.assertEqual(branch7, branch_no_checkout)
+    def tearDown(self):
+        self.patch_workdir.stop()
+        self.patch_agentdir.stop()
+        self.mock_ctx.reset()
 
-#     def test_context_no_checkout_error(self):
-#         ctx = get_ctx()
+    def test_context_is_worktree_true(self):
+        with agent_context(self.mock_ctx, '6.53.x'):
+            self.assertTrue(is_worktree())
 
-#         with agent_context(ctx, '6.53.x'):
-#             pass
+    def test_context_is_worktree_false(self):
+        self.assertFalse(is_worktree())
 
-#         def switch_context():
-#             # The current branch is not main
-#             with agent_context(ctx, 'main', skip_checkout=True):
-#                 pass
+    def test_context_nested(self):
+        with agent_context(self.mock_ctx, '6.53.x'):
+            with agent_context(self.mock_ctx, '6.53.x'):
+                self.assertTrue(is_worktree())
+            self.assertTrue(is_worktree())
 
-#         self.assertRaises(AssertionError, switch_context)
+    def test_context_pwd(self):
+        ctx = get_ctx()
+
+        with agent_context(self.mock_ctx, None, skip_checkout=True):
+            pwdnone = ctx.run('pwd').stdout
+
+        with agent_context(self.mock_ctx, '6.53.x'):
+            pwd6 = ctx.run('pwd').stdout
+
+        with agent_context(self.mock_ctx, 'main'):
+            pwdmain = ctx.run('pwd').stdout
+
+        self.assertEqual(pwd6, pwdnone)
+        self.assertEqual(pwd6, pwdmain)
+
+    def test_context_modules(self):
+        with agent_context(self.mock_ctx, 'main'):
+            modules7 = get_default_modules()
+
+        with agent_context(self.mock_ctx, '6.53.x'):
+            modules6 = get_default_modules()
+
+        self.assertNotEqual(set(modules6.keys()), set(modules7.keys()))
+
+    def test_context_no_checkout(self):
+        with agent_context(self.mock_ctx, 'main'):
+            modules7 = get_default_modules()
+
+        with agent_context(self.mock_ctx, '6.53.x'):
+            modules6 = get_default_modules()
+
+        # Cannot skip checkout if the branch is not the target one (current is 6.53.x)
+        self.assertRaises(AssertionError, lambda: agent_context(self.mock_ctx, 'main', skip_checkout=True).__enter__())
+
+        with agent_context(self.mock_ctx, '6.53.x', skip_checkout=True):
+            modules_no_checkout = get_default_modules()
+
+        self.assertNotEqual(set(modules6.keys()), set(modules7.keys()))
+        self.assertEqual(set(modules6.keys()), set(modules_no_checkout.keys()))