Merge pull request #2 from hardcoretech/feat/add_git_awareness

feature(mode): Add git awareness modes

.github/workflows/ut.yml

@@ -0,0 +1,25 @@
+name: Run Unit Test
+
+on: [pull_request]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+        with:
+          python-version: "3.6.8"
+      - name: Install dependencies
+        run: |
+          pip install pip --upgrade
+          pip install pytest
+      - name: Test with pytest
+        run: |
+          PYTHONPATH=src pytest --log-level=debug --junitxml=junit_report.xml .
+      - name: Publish Unit Test Results
+        uses: EnricoMi/publish-unit-test-result-action/composite@v1
+        if: always()
+        with:
+          files: junit_report.xml

src/missing.py

@@ -1,19 +1,33 @@
 #! /usr/bin/env python
 import argparse
+import enum
 import logging
 import os
 import re
+import subprocess
+from functools import lru_cache
 from typing import Optional
 
 logger = logging.getLogger('missing')
 
+class Mode(enum.Enum):
+    ALL = 'all'
+    OBEY_GITIGNORE = 'obey_gitignore'
+    STAGED_ONLY = 'staged_only'
+
+    def __str__(self):
+        return self.value
+
 class Missing:
     RE_TEST_PY = re.compile(r'^test.*?\.py$')
 
-    def __init__(self, exclude):
+    def __init__(self, mode, exclude):
         if not (isinstance(exclude, list) or isinstance(exclude, tuple)):
             raise TypeError('exclude should be list or tuple')
 
+        self._tracked_files = self._find_tracked_files(mode)
+        logger.debug('Tracked files: %r', self._tracked_files)
+
         # append .git to exclude folder
         exclude = set(['.git', *exclude])
 
@@ -28,8 +42,9 @@ def run(self) -> int:
 
         return fail
 
+    @lru_cache(maxsize=None)
     def _check_init_py_exists(self, path: str) -> bool:
-        '''check the __init__.py exists on the current path and parent path'''
+        """check the __init__.py exists on the current path and parent path"""
         fail = False
         basedir = os.path.dirname(path)
 
@@ -43,14 +58,14 @@ def _check_init_py_exists(self, path: str) -> bool:
         init_py = f'{basedir}/__init__.py'
         if not os.path.exists(init_py):
             fail = True
+
             # create the __init__.py
             with open(init_py, 'w') as fd:
                 logger.warning('create file: {}'.format(init_py))
-
         return fail
 
     def _find_for_unittest(self, basedir: Optional[str] = None) -> bool:
-        '''the unittest find the test*.py only for the regular package'''
+        """the unittest find the test*.py only for the regular package"""
         fail = False
         init_run = False
 
@@ -62,25 +77,102 @@ def _find_for_unittest(self, basedir: Optional[str] = None) -> bool:
             path = os.path.normpath(f'{basedir}/{filename}')
             if path in self._exclude:
                 # skip the explicitly excluded path
+                logger.debug('exclude: %r', path)
                 continue
 
             if os.path.isdir(path):
                 if self._find_for_unittest(path):
                     fail = True
             elif self.RE_TEST_PY.match(filename):
-                if self._check_init_py_exists(path):
-                    fail = True
+                if self._is_tracked(path):
+                    logger.debug('check: %r', path)
+                    if self._check_init_py_exists(path):
+                        fail = True
+                else:
+                    logger.debug('untracked: %r', path)
+            else:
+                logger.debug('no_match: %r', path)
 
         if init_run and fail:
             logger.warning('found missing __init__.py for unittest')
 
         return fail
 
+    def _is_tracked(self, path):
+        if self._tracked_files is None:
+            return True
+        return path in self._tracked_files
+
+    def _find_tracked_files(self, mode):
+        if Mode(mode) == Mode.OBEY_GITIGNORE:
+            return self._list_files_obey_gitignore()
+        elif Mode(mode) == Mode.STAGED_ONLY:
+            return self._list_files_staged_only()
+        else:
+            return None
+
+    def _list_files_staged_only(self):
+        # list staged files
+        tracked_files = (
+            subprocess.check_output(
+                ['git', '--no-pager', 'diff', '--name-only', '--cached', '-z'],
+                encoding='utf-8',
+            )
+            .rstrip('\0')
+            .split('\0')
+        )
+        return set(
+            [
+                os.path.normpath(tracked_file)
+                for tracked_file in tracked_files
+                if self.RE_TEST_PY.match(os.path.basename(tracked_file))
+            ]
+        )
+
+    def _list_files_obey_gitignore(self):
+        # list committed files
+        tracked_files = (
+            subprocess.check_output(
+                ['git', 'ls-files', '-z'],
+                encoding='utf-8',
+            )
+            .rstrip('\0')
+            .split('\0')
+        )
+
+        # list untracked files
+        tracked_files += (
+            subprocess.check_output(
+                ['git', 'ls-files', '-o', '--exclude-standard', '-z'],
+                encoding='utf-8',
+            )
+            .rstrip('\0')
+            .split('\0')
+        )
+        return set(
+            [
+                os.path.normpath(tracked_file)
+                for tracked_file in tracked_files
+                if self.RE_TEST_PY.match(os.path.basename(tracked_file))
+            ]
+        )
+
 
 def main() -> int:
     parser = argparse.ArgumentParser(description='Find the missing but necessary files')
     parser.add_argument('-e', '--exclude', nargs='*', help='exclude the file or folder')
-    parser.add_argument('-q', '--quite', action='store_true', default=False, help='disable all log')
+    parser.add_argument(
+        '-q', '--quite', action='store_true', default=False, help='disable all log'
+    )
+
+
+    parser.add_argument(
+        '-m',
+        '--mode',
+        type=Mode,
+        default=Mode.ALL,
+        choices=list(Mode)
+    )
     args = parser.parse_args()
 
     if args.quite:
@@ -91,8 +183,9 @@ def main() -> int:
             handler = logging.StreamHandler()
             logger.addHandler(handler)
 
-    missing = Missing(args.exclude or [])
+    missing = Missing(args.mode, args.exclude or [])
     return missing.run()
 
+
 if __name__ == '__main__':
     exit(main())

tests/test_missing.py

@@ -0,0 +1,69 @@
+import os
+import tempfile
+
+import pytest
+
+from missing import Missing, Mode
+
+
+@pytest.fixture
+def temp_git_folder():
+    # Create a temp git folder so we could conduct the integration test
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        os.chdir(tmp_dir)
+        with open('.gitignore', 'w+') as fout:
+            fout.write('test_ignore.py')
+
+        os.mkdir('tests')
+        os.mkdir(os.path.join('tests', 'ignored'))
+        with open(os.path.join('tests', 'ignored', 'test_ignore.py'), 'w+'):
+            pass
+
+        os.mkdir(os.path.join('tests', 'untracked'))
+        with open(os.path.join('tests', 'untracked', 'test_untracked.py'), 'w+'):
+            pass
+
+        os.mkdir(os.path.join('tests', 'staged'))
+        with open(os.path.join('tests', 'staged', 'test_staged.py'), 'w+'):
+            pass
+
+        os.mkdir('data')
+        with open(os.path.join('data', 'user.xml'), 'w+') as fount:
+            pass
+
+        os.system('git init')
+        os.system('git add %s' % os.path.join('tests', 'staged'))
+        yield
+
+
+class TestMissing:
+    def test__default_mode(self, temp_git_folder):
+        assert 1 == Missing(mode='all', exclude=[]).run()
+        assert not os.path.exists('data/__init__.py')
+        assert os.path.isfile('tests/ignored/__init__.py')
+        assert os.path.isfile('tests/untracked/__init__.py')
+        assert os.path.isfile('tests/staged/__init__.py')
+        assert os.path.isfile('tests/__init__.py')
+        assert 0 == Missing(mode=Mode.ALL, exclude=[]).run()
+
+    def test__obey_gitignore(self, temp_git_folder):
+        assert 1 == Missing(mode='obey_gitignore', exclude=[]).run()
+        assert not os.path.exists('data/__init__.py')
+        assert not os.path.exists('tests/ignored/__init__.py')
+        assert os.path.isfile('tests/untracked/__init__.py')
+        assert os.path.isfile('tests/staged/__init__.py')
+        assert os.path.isfile('tests/__init__.py')
+        assert 0 == Missing(mode=Mode.OBEY_GITIGNORE, exclude=[]).run()
+
+    def test__staged_only(self, temp_git_folder):
+        assert 1 == Missing(mode='staged_only', exclude=[]).run()
+        assert not os.path.exists('data/__init__.py')
+        assert not os.path.exists('tests/ignored/__init__.py')
+        assert not os.path.exists('tests/untracked/__init__.py')
+        assert os.path.isfile('tests/staged/__init__.py')
+        assert os.path.isfile('tests/__init__.py')
+        assert 0 == Missing(mode=Mode.STAGED_ONLY, exclude=[]).run()
+
+    def test__exclude(self, temp_git_folder):
+        assert 0 == Missing(mode=Mode.ALL, exclude=['tests']).run()