Merge pull request #15 from walefy/test-initials

test: write some testes

.coveragerc

@@ -2,4 +2,4 @@
 omit = tests/*,db/database.py,__init__.py
 
 [report]
-show_missing = True
+show_missing = True

.github/workflows/tests.yml

@@ -10,8 +10,13 @@ jobs:
   tests:
     runs-on: ubuntu-latest
 
+    permissions:
+      pull-requests: write
+
     steps:
-      - uses: actions/checkout@v2
+      - name: Fetch project code
+        uses: actions/checkout@v2
+
       - name: Set up Python 3.11.4
         uses: actions/setup-python@v2
         with:
@@ -26,5 +31,15 @@ jobs:
       - name: Run Flake8
         run: poetry run flake8 .
 
-      - name: Run tests with pytest
+      - name: Run test
         run: poetry run task test
+
+      - name: Generate report
+        run: poetry run task report-md
+        if: always()
+
+      - name: Commet Report PR
+        uses: thollander/actions-comment-pull-request@v2
+        if: always()
+        with:
+          filePath: ./report.md

main.py

@@ -1,14 +1,13 @@
 from os import path
 from shutil import rmtree
 
-import xmltodict
 from fastapi import FastAPI, Header, HTTPException, Request, UploadFile, status
 from fastapi.responses import JSONResponse
 
 from crud import insert_nfe
 from db.database import init_db
 from models.company import Company, CompanyRegistration
-from utils import read_all_xml_files, unzip_file, Logger
+from utils import Logger, read_all_xml_files, unzip_file
 from validations import (
     check_duplicates,
     company_exists,
@@ -83,11 +82,7 @@ async def register_company(company_registration: CompanyRegistration):
 async def xml_interpreter(upload_file: UploadFile = None, cnpj: str = Header(...)):
     try:
         if upload_file is None:
-            raise HTTPException(status_code=400, detail='xml file not found!')
-
-        if upload_file.filename.endswith('.xml'):
-            doc = xmltodict.parse(upload_file.file.read())
-            return doc
+            raise HTTPException(status_code=400, detail={'detail': 'xml file not found!'})
 
         if upload_file.filename.endswith('.zip'):
             response_dict = {
@@ -117,7 +112,7 @@ async def xml_interpreter(upload_file: UploadFile = None, cnpj: str = Header(...
             return response_dict
 
         else:
-            raise HTTPException(status_code=400, detail='xml file not found!')
+            raise HTTPException(status_code=400, detail={'detail': 'file must be a zip!'})
 
     except HTTPException as http_error:
         raise http_error

pyproject.toml

@@ -10,9 +10,11 @@ multi_line_output = 3
 
 [tool.taskipy.tasks]
 coverage = "pytest -W ignore::DeprecationWarning --cov=./ --cov-config=.coveragerc tests/ && rm .coverage"
+coverage-html = "pytest -W ignore::DeprecationWarning --cov=./ --cov-config=.coveragerc tests/ && coverage html && rm -rf .coverage"
 isort = "isort ."
 lint = "flake8"
-test = "pytest -W ignore::DeprecationWarning"
+report-md = "pytest -W ignore::DeprecationWarning -vv > report.txt ; cat report.txt ; python ./tests/helpers/generate_report.py ; rm report.txt"
+test = "pytest -W ignore::DeprecationWarning -vv"
 
 [tool.poetry.dependencies]
 beanie = "^1.22.6"
@@ -28,7 +30,7 @@ flake8 = "^6.1.0"
 httpx = "^0.25.0"
 isort = "^5.12.0"
 mongomock-motor = "^0.0.21"
-pytest = "^7.4.2"
+pytest = "^7.4.3"
 pytest-asyncio = "^0.21.1"
 pytest-cov = "^4.1.0"
 taskipy = "^1.12.2"

tests/helpers/generate_report.py

@@ -0,0 +1,38 @@
+from typing import Literal
+
+
+class TestItem:
+    passed: bool
+    name: str
+
+    def __init__(self, name: str, passed: Literal['passed', 'failed']):
+        self.name = name
+        self.passed = passed == 'passed'
+
+    def __str__(self) -> str:
+        return f' TestItem (name = {self.name}, passed = {self.passed})'
+
+
+def generate_report():
+    tests: list[TestItem] = []
+
+    with open('report.txt', 'r') as file:
+        for line in file.readlines():
+            if line.startswith('test'):
+                test = line.split(' ')
+                test_name = test[0].split('::')[1]
+                test_passed = test[1].lower()
+                test = TestItem(test_name, test_passed)
+                tests.append(test)
+
+            if line.strip().endswith('[100%]'):
+                break
+
+    with open('report.md', 'w') as file:
+        file.write('*name* | *passed*\n')
+        file.write('--- | :---:\n')
+        for test in tests:
+            file.write(f'{test.name} | {"✅" if test.passed else "❌"}\n')
+
+
+generate_report()

tests/integration/send_zip_file_test.py

@@ -1,5 +1,6 @@
 import unittest
 from os import path
+from unittest.mock import Mock, patch
 
 import pytest
 import pytest_asyncio
@@ -82,7 +83,7 @@ async def test_send_zip_file_with_gap():
 
         response = await async_client.post('/xml', files=files, headers=headers)
         zip_file.close()
-    
+
     expected_warnings = {
         'warnings': [
             {
@@ -96,9 +97,133 @@ async def test_send_zip_file_with_gap():
             },
         ],
     }
-    
+
     actual_warnings = response.json()
     case = unittest.TestCase()
 
     case.assertCountEqual(actual_warnings, expected_warnings)
     assert response.status_code == status.HTTP_201_CREATED
+
+
+@pytest.mark.asyncio
+async def test_make_request_without_upload_file():
+    async with AsyncClient(app=app, base_url='http://test') as async_client:
+        headers = {
+            'Content-Type': 'multipart/form-data;boundary=boundary',
+            'cnpj': '40028176000176'
+        }
+
+        response = await async_client.post('/xml', headers=headers)
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json() == {'detail': 'xml file not found!'}
+
+
+@pytest.mark.asyncio
+async def test_make_request_without_company_registered():
+    compare_cnpj_in_all_files = Mock(return_value=None)
+    verify_sequence_with_gap = Mock(return_value=None)
+
+    patcher_compare_cnpj = patch(
+        'main.compare_cnpj_in_all_files', compare_cnpj_in_all_files)
+    patcher_verify_sequence = patch(
+        'main.verify_sequence_with_gap', verify_sequence_with_gap)
+
+    patcher_compare_cnpj.start()
+    patcher_verify_sequence.start()
+
+    async with AsyncClient(app=app, base_url='http://test') as async_client:
+        path_to_zip_file = path.join(
+            path.dirname(__file__),
+            '..',
+            'data',
+            'minimal_no_gap.zip'
+        )
+
+        zip_file = open(path_to_zip_file, 'rb')
+
+        files = {
+            'upload_file': zip_file
+        }
+
+        headers = {
+            'Content-Type': 'multipart/form-data;boundary=boundary',
+            'cnpj': '40028176000177'
+        }
+
+        response = await async_client.post('/xml', files=files, headers=headers)
+        zip_file.close()
+
+    patcher_compare_cnpj.stop()
+    patcher_verify_sequence.stop()
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json() == {'message': 'Company not registered!'}
+
+
+@pytest.mark.asyncio
+async def test_make_request_with_invalid_file_extension():
+    async with AsyncClient(app=app, base_url='http://test') as async_client:
+        path_to_zip_file = path.join(
+            path.dirname(__file__),
+            '..',
+            'data',
+            'full_no_gap',
+            '63.xml'
+        )
+
+        zip_file = open(path_to_zip_file, 'rb')
+
+        files = {
+            'upload_file': zip_file
+        }
+
+        headers = {
+            'Content-Type': 'multipart/form-data;boundary=boundary',
+            'cnpj': '40028176000176'
+        }
+
+        response = await async_client.post('/xml', files=files, headers=headers)
+        zip_file.close()
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json() == {'detail': 'file must be a zip!'}
+
+
+@pytest.mark.asyncio
+async def test_make_request_with_different_cnpj():
+    async with AsyncClient(app=app, base_url='http://test') as async_client:
+        path_to_zip_file = path.join(
+            path.dirname(__file__),
+            '..',
+            'data',
+            'minimal_no_gap.zip'
+        )
+
+        zip_file = open(path_to_zip_file, 'rb')
+
+        files = {
+            'upload_file': zip_file
+        }
+
+        headers = {
+            'Content-Type': 'multipart/form-data;boundary=boundary',
+            'cnpj': '40028176000177'
+        }
+
+        response = await async_client.post('/xml', files=files, headers=headers)
+        zip_file.close()
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()['message'] == 'The CNPJ is not match!'
+
+    not_match_files_expect = [
+        '63.xml',
+        '64.xml',
+        '65.xml',
+        '66.xml'
+    ]
+    not_match_files_actual = response.json()['files']
+
+    case = unittest.TestCase()
+    case.assertCountEqual(not_match_files_actual, not_match_files_expect)

tests/unit/get_nested_value_test.py

@@ -0,0 +1,23 @@
+import pytest
+
+from utils import get_nested_value
+
+
+@pytest.mark.asyncio
+async def test_get_nested_value():
+    list_key = ('a', 'b', 'c')
+    entry_dict = {'a': {'b': {'c': 'test'}}}
+
+    assert get_nested_value(list_key, entry_dict) == 'test'
+
+
+@pytest.mark.asyncio
+async def test_get_nested_value_with_nonexistent_key():
+    list_key = ('a', 'b', 'd')
+    dict_test = {'a': {'b': {'c': 'cheguei'}}}
+
+    with pytest.raises(ValueError):
+        get_nested_value(list_key, dict_test)
+
+    with pytest.raises(ValueError):
+        get_nested_value(list_key, {'a': {'b': 'c'}})

tests/unit/http_exception_handler_test.py

@@ -1,7 +1,9 @@
-import pytest
 from unittest.mock import Mock, patch
-from main import http_exception_handler
+
+import pytest
 from fastapi import HTTPException, Request
+
+from main import http_exception_handler
 from utils import Logger
 
 

tests/unit/insert_nfe_test.py

@@ -1,6 +1,7 @@
+from unittest.mock import AsyncMock, Mock
+
 import pytest
 import pytest_asyncio
-from unittest.mock import AsyncMock, Mock
 
 from crud import insert_nfe
 from models.company import Company
@@ -13,6 +14,7 @@ async def mock_mongodb():
     document_models = [Company]
     await init_mock_mongodb_beanie(document_models)
 
+
 @pytest.mark.asyncio
 async def test_insert_nfe_with_multiple_products():
     list_nfe_json = [
@@ -22,8 +24,12 @@ async def test_insert_nfe_with_multiple_products():
                     'NFe': {
                         'infNFe': {
                             'det': [
-                                {'prod': {'xProd': 'Product 1', 'vProd': '10.00'}, 'imposto': {}},
-                                {'prod': {'xProd': 'Product 2', 'vProd': '20.00'}, 'imposto': {}},
+                                {'prod': {'xProd': 'Product 1', 'vProd': '10.00'},
+                                 'imposto': {}
+                                 },
+                                {'prod': {'xProd': 'Product 2', 'vProd': '20.00'},
+                                 'imposto': {}
+                                 },
                             ],
                             'ide': {'nNF': '123', 'serie': '1'},
                         }
@@ -41,4 +47,4 @@ async def test_insert_nfe_with_multiple_products():
 
     await insert_nfe('40028176000176', list_nfe_json)
 
-    company.update.assert_called_once()
+    company.update.assert_called_once()

tests/unit/logger_test.py

@@ -0,0 +1,30 @@
+import tempfile
+from os import path
+
+import pytest
+
+from utils import Logger
+
+
+@pytest.mark.asyncio
+async def test_logger_class():
+    temp_folder = tempfile.mkdtemp('logs')
+
+    logger = Logger(folder=temp_folder)
+    logger.log('test')
+
+    with open(logger.path_to_file, 'r') as file:
+        assert file.read() == 'test\n'
+
+
+@pytest.mark.asyncio
+async def test_logger_class_with_nonexistent_folder():
+    temp_folder = tempfile.mkdtemp('test')
+
+    logger = Logger(folder=temp_folder + '/logs')
+    logger.log('test')
+
+    with open(logger.path_to_file, 'r') as file:
+        assert file.read() == 'test\n'
+
+    assert path.exists(temp_folder + '/logs')