Merge branch 'pytest', remote-tracking branch 'juanpsenn/revalidate_r…

…eceipt'

PRs #95 and #89

CHANGELOG.rst

@@ -16,6 +16,8 @@ acá.
 * Soporte para Zeep ~4.0.0.
 * Soporte para Django 3.2.
 * La [mayoría de la] documentación ahora está traducida.
+* Se implementa :meth:`~.Receipt.revalidate`. Provee un mecanismo de revalidacion
+  de un comprobante para completar datos faltandes referentes a la validacion del mismo.
 
 8.0.4
 -----

django_afip/factories.py

@@ -6,6 +6,7 @@
 from django.utils.timezone import make_aware
 from factory import LazyFunction
 from factory import PostGenerationMethodCall
+from factory import Sequence
 from factory import SubFactory
 from factory import post_generation
 from factory.django import DjangoModelFactory
@@ -152,10 +153,9 @@ def post(obj: models.Receipt, create, extracted, **kwargs):
         TaxFactory(tax_type__code=3, receipt=obj)
 
 
-class ReceiptWithInconsistentVatAndTaxFactory(ReceiptFactory):
+class ReceiptWithInconsistentVatAndTaxFactory(ReceiptWithVatAndTaxFactory):
     """Receipt with a valid Vat and Tax, ready to validate."""
 
-    point_of_sales = LazyFunction(lambda: models.PointOfSales.objects.first())
     document_type = SubFactory(DocumentTypeFactory, code=80)
 
     @post_generation
@@ -164,6 +164,16 @@ def post(obj: models.Receipt, create, extracted, **kwargs):
         TaxFactory(tax_type__code=3, receipt=obj)
 
 
+class ReceiptWithApprovedValidation(ReceiptFactory):
+    """Receipt with fake (e.g.: not live) approved validation."""
+
+    receipt_number = Sequence(lambda n: n + 1)
+
+    @post_generation
+    def post(obj: models.Receipt, create, extracted, **kwargs):
+        ReceiptValidationFactory(receipt=obj)
+
+
 class ReceiptValidationFactory(DjangoModelFactory):
     class Meta:
         model = models.ReceiptValidation
@@ -190,6 +200,10 @@ class Meta:
     vat_condition = "Responsable Monotributo"
 
 
+class ReceiptPDFWithFileFactory(ReceiptPDFFactory):
+    receipt = SubFactory(ReceiptWithApprovedValidation)
+
+
 class GenericAfipTypeFactory(DjangoModelFactory):
     class Meta:
         model = models.GenericAfipType

django_afip/models.py

@@ -8,6 +8,7 @@
 from io import BytesIO
 from tempfile import NamedTemporaryFile
 from typing import List
+from typing import Optional
 from typing import Type
 from uuid import uuid4
 
@@ -889,6 +890,10 @@ def fetch_last_receipt_number(self, point_of_sales, receipt_type):
 
     def fetch_receipt_data(self, receipt_type, receipt_number, point_of_sales):
         """Returns receipt related data"""
+
+        if not receipt_number:
+            return None
+
         client = clients.get_client("wsfe", point_of_sales.owner.is_sandboxed)
         response_xml = client.service.FECompConsultar(
             serializers.serialize_ticket(
@@ -898,9 +903,11 @@ def fetch_receipt_data(self, receipt_type, receipt_number, point_of_sales):
                 receipt_type, receipt_number, point_of_sales.number
             ),
         )
-        check_response(response_xml)
-
-        return response_xml.ResultGet
+        try:
+            check_response(response_xml)
+            return response_xml.ResultGet
+        except exceptions.AfipException:
+            return None
 
     def get_queryset(self):
         return ReceiptQuerySet(self.model, using=self._db).select_related(
@@ -1116,6 +1123,50 @@ def validate(self, ticket=None, raise_=False):
             raise exceptions.ValidationError(rv[0])
         return rv
 
+    def revalidate(self) -> Optional["ReceiptValidation"]:
+        """Revalidate this receipt.
+
+        Fetches data of a validated receipt from AFIP's servers.
+        If the receipt exists a ``ReceiptValidation`` instance is
+        created and returned, otherwise, returns ``None``.
+        If there is already a ``ReceiptValidation`` for this instance,
+        returns ``self.validation``.
+        This should be used for verification purpose, here's a list of
+        some use cases:
+         - Incomplete validation process
+         - Fetch CAE data from AFIP's servers
+        """
+        # This may avoid unnecessary revalidation
+        if self.is_validated:
+            return self.validation
+
+        receipt_data = Receipt.objects.fetch_receipt_data(
+            self.receipt_type.code, self.receipt_number, self.point_of_sales
+        )
+
+        if not receipt_data:
+            return None
+
+        if receipt_data.Resultado == ReceiptValidation.RESULT_APPROVED:
+            validation = ReceiptValidation.objects.create(
+                result=receipt_data.Resultado,
+                cae=receipt_data.CodAutorizacion,
+                cae_expiration=parsers.parse_date(receipt_data.FchVto),
+                receipt=self,
+                processed_date=parsers.parse_datetime(
+                    receipt_data.FchProceso,
+                ),
+            )
+            if receipt_data.Observaciones:
+                for obs in receipt_data.Observaciones.Obs:
+                    observation = Observation.objects.get_or_create(
+                        code=obs.Code,
+                        message=obs.Msg,
+                    )
+                validation.observations.add(observation)
+            return validation
+        return None
+
     def __repr__(self):
         return "<Receipt {}: {} {} for {}>".format(
             self.pk,

testapp/conftest.py

@@ -1,6 +1,24 @@
 import pytest
+from django.conf import settings
+from django.core import serializers
 
+from django_afip import models
+from django_afip.exceptions import AuthenticationError
+from django_afip.factories import TaxPayerFactory
 from django_afip.factories import get_test_file
+from django_afip.models import AuthTicket
+
+CACHED_TICKET_PATH = settings.BASE_DIR / "test_ticket.yaml"
+_live_mode = False
+
+
+def pytest_runtest_setup(item):
+    """Set live mode if the marker has been passed to pytest.
+
+    This avoid accidentally using any of the live-mode fixtures in non-live mode."""
+    if list(item.iter_markers(name="live")):
+        global _live_mode
+        _live_mode = True
 
 
 @pytest.fixture
@@ -13,3 +31,65 @@ def expired_crt() -> bytes:
 def expired_key() -> bytes:
     with open(get_test_file("test_expired.key"), "rb") as key:
         return key.read()
+
+
+@pytest.fixture
+def live_taxpayer(db):
+    """Return a taxpayer usable with AFIP's test servers."""
+    return TaxPayerFactory(pk=1)
+
+
+@pytest.fixture
+def live_ticket(db, live_taxpayer):
+    """Return an authentication ticket usable with AFIP's test servers.
+
+    AFIP doesn't allow requesting tickets too often, so we after a few runs
+    of the test suite, we can't generate tickets any more and have to wait.
+
+    This helper generates a ticket, and saves it to disk into the app's
+    BASE_DIR, so that developers can run tests over and over without having to
+    worry about the limitation.
+
+    Expired tickets are not deleted or handled properly; it's up to you to
+    delete stale cached tickets.
+
+    When running in CI pipelines, this file will never be preset so won't be a
+    problem.
+    """
+    assert _live_mode
+
+    # Try reading a cached ticket from disk:
+    try:
+        with open(CACHED_TICKET_PATH) as f:
+            [obj] = serializers.deserialize("yaml", f.read())
+            obj.save()
+
+        return obj
+    except FileNotFoundError:
+        # If something failed, we should still have no tickets in the DB:
+        assert models.AuthTicket.objects.count() == 0
+
+    # If nothing cached, create a new one:
+    try:
+        ticket = AuthTicket.objects.get_any_active("wsfe")
+    except AuthenticationError as e:
+        pytest.exit(f"Bailing due to failure authenticating with AFIP:\n{e}")
+
+    # We got a ticket, and it should be saved to DB:
+    assert models.AuthTicket.objects.count() == 1
+
+    # TODO: Somehow detect if the ticket has expired?
+
+    data = serializers.serialize("yaml", [ticket])
+    with open(CACHED_TICKET_PATH, "w") as f:
+        f.write(data)
+
+    return ticket
+
+
+@pytest.fixture
+def populated_db(db, live_ticket, live_taxpayer):
+    """Populate the database with fixtures and a POS"""
+
+    models.load_metadata()
+    live_taxpayer.fetch_points_of_sales()

testapp/testapp/settings.py

@@ -1,9 +1,10 @@
 import os
+from pathlib import Path
 
 import dj_database_url
 
 # Build paths inside the project like this: os.path.join(BASE_DIR, ...)
-BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+BASE_DIR = Path(__file__).resolve().parent.parent
 
 
 # Quick-start development settings - unsuitable for production

testapp/testapp/testmain/tests/test_admin.py

@@ -231,16 +231,13 @@ def test_receipt_admin_get_exclude():
 
 
 @pytest.mark.django_db
-def create_test_receipt_pdfs():
-    validation = factories.ReceiptValidationFactory()
-    with_file = factories.ReceiptPDFFactory(receipt=validation.receipt)
+def test_receipt_pdf_factories_and_files():
+    with_file = factories.ReceiptPDFWithFileFactory()
     without_file = factories.ReceiptPDFFactory()
 
     assert not without_file.pdf_file
     assert with_file.pdf_file
 
-    return with_file, without_file
-
 
 def test_has_file_filter_all(admin_client):
     """Check that the has_file filter applies properly
@@ -249,23 +246,26 @@ def test_has_file_filter_all(admin_client):
     object's change page is present, since no matter how we reformat the rows,
     this will always be present as long as the object is listed.
     """
-    with_file, without_file = create_test_receipt_pdfs()
+    with_file = factories.ReceiptPDFWithFileFactory()
+    without_file = factories.ReceiptPDFFactory()
 
     response = admin_client.get("/admin/afip/receiptpdf/")
     assertContains(response, f"/admin/afip/receiptpdf/{with_file.pk}/change/")
     assertContains(response, f"/admin/afip/receiptpdf/{without_file.pk}/change/")
 
 
 def test_has_file_filter_with_file(admin_client):
-    with_file, without_file = create_test_receipt_pdfs()
+    with_file = factories.ReceiptPDFWithFileFactory()
+    without_file = factories.ReceiptPDFFactory()
 
     response = admin_client.get("/admin/afip/receiptpdf/?has_file=yes")
     assertContains(response, f"/admin/afip/receiptpdf/{with_file.pk}/change/")
     assertNotContains(response, f"/admin/afip/receiptpdf/{without_file.pk}/change/")
 
 
 def test_has_file_filter_without_file(admin_client):
-    with_file, without_file = create_test_receipt_pdfs()
+    with_file = factories.ReceiptPDFWithFileFactory()
+    without_file = factories.ReceiptPDFFactory()
 
     response = admin_client.get("/admin/afip/receiptpdf/?has_file=no")
     assertNotContains(response, f"/admin/afip/receiptpdf/{with_file.pk}/change/")