vbv/server/integration_tests/abacus_sftp/test_abacus_sftp.py

import os
import shutil
import tempfile
from datetime import date
from io import StringIO
from subprocess import Popen
from time import sleep

import pytest
from django.conf import settings
from freezegun import freeze_time

from vbv_lernwelt.core.admin import User
from vbv_lernwelt.core.create_default_users import create_default_users
from vbv_lernwelt.core.model_utils import add_countries
from vbv_lernwelt.shop.invoice.abacus import abacus_ssh_upload
from vbv_lernwelt.shop.invoice.abacus_sftp_client import AbacusSftpClient
from vbv_lernwelt.shop.tests.factories import CheckoutInformationFactory


@pytest.fixture(scope="module")
def sftp_server():
    tmppath = tempfile.mkdtemp()
    print(tmppath)
    shutil.rmtree(tmppath)
    os.mkdir(tmppath)
    os.mkdir(os.path.join(tmppath, "debitor"))
    os.mkdir(os.path.join(tmppath, "order"))
    sftp_server = Popen(
        f"sftpserver -p {settings.ABACUS_EXPORT_SFTP_PORT} -l INFO",
        shell=True,
        cwd=tmppath,
    )
    sleep(3)
    yield tmppath
    if sftp_server:
        sftp_server.kill()


def test_can_write_file_to_fake_sftp_server(sftp_server):
    with AbacusSftpClient() as client:
        files = client.listdir(".")
        assert set(files) == {"debitor", "order"}

        str_file = StringIO()
        str_file.write("Hello world\n")
        str_file.seek(0)
        client.putfo(str_file, "hello.txt")

        files = client.listdir(".")
        assert set(files) == {"debitor", "order", "hello.txt"}


@pytest.fixture
def setup_abacus_env(sftp_server):
    add_countries(small_set=True)
    create_default_users()
    yield sftp_server


@pytest.mark.django_db
def test_upload_abacus_xml(setup_abacus_env):
    tmppath = setup_abacus_env

    # set abacus_number before
    _pat = User.objects.get(username="patrizia.huggel@eiger-versicherungen.ch")
    _pat.abacus_debitor_number = 60000011
    _pat.save()
    _ignore_checkout_information = CheckoutInformationFactory(
        user=_pat, abacus_order_id=6_000_000_123
    )

    feuz = User.objects.get(username="andreas.feuz@eiger-versicherungen.ch")

    with freeze_time("2024-02-15 08:33:12"):
        feuz_checkout_info = CheckoutInformationFactory(
            user=feuz,
            transaction_id="24021508331287484",
            first_name="Andreas",
            last_name="Feuz",
            street="Eggersmatt",
            street_number="32",
            postal_code="1719",
            city="Zumholz",
            country_id="CH",
            invoice_address="org",
            organisation_detail_name="VBV",
            organisation_street="Laupenstrasse",
            organisation_street_number="10",
            organisation_postal_code="3000",
            organisation_city="Bern",
            organisation_country_id="CH",
        )

    with freeze_time("2024-02-15 09:37:41"):
        abacus_ssh_upload(feuz_checkout_info)

        # check transmission data
        feuz_checkout_info.refresh_from_db()
        assert feuz_checkout_info.invoice_transmitted_at.date() == date(2024, 2, 15)
        abacus_transmission_list = feuz_checkout_info.additional_json_data.get(
            "abacus_transmission_list", []
        )
        assert len(abacus_transmission_list) == 1
        assert abacus_transmission_list[0]["transmitted_at"][0:10] == "2024-02-15"

        # check if files got created
        debitor_filepath = os.path.join(tmppath, "debitor/myVBV_debi_60000012.xml")
        assert os.path.exists(debitor_filepath)

        with open(debitor_filepath) as debitor_file:
            debi_content = debitor_file.read()
            assert "<CustomerNumber>60000012</CustomerNumber>" in debi_content
            assert "<Email>andreas.feuz@eiger-versicherungen.ch</Email>" in debi_content

        order_filepath = os.path.join(
            tmppath, "order/myVBV_orde_20240215083312_60000012_6000000124.xml"
        )
        assert os.path.exists(order_filepath)
        with open(order_filepath) as order_file:
            order_content = order_file.read()
            assert (
                "<ReferencePurchaseOrder>24021508331287484</ReferencePurchaseOrder>"
                in order_content
            )
            assert "<CustomerNumber>60000012</CustomerNumber>" in order_content

        feuz_checkout_info.refresh_from_db()
        assert feuz_checkout_info.abacus_ssh_upload_done

        # calling `abacus_ssh_upload` a second time will not upload files again...
        os.remove(debitor_filepath)
        os.remove(order_filepath)

        abacus_ssh_upload(feuz_checkout_info)

        debitor_filepath = os.path.join(tmppath, "debitor/myVBV_debi_60000012.xml")
        assert not os.path.exists(debitor_filepath)

        order_filepath = os.path.join(
            tmppath, "order/myVBV_orde_60000012_20240215083312_6000000124.xml"
        )
        assert not os.path.exists(order_filepath)

    with freeze_time("2024-04-21 17:33:19"):
        # change customer data later will retrigger and redo abacus upload
        feuz_checkout_info.first_name = "Peter"
        feuz_checkout_info.last_name = "Egger"
        feuz_checkout_info.save()

        abacus_ssh_upload(feuz_checkout_info)

        # check transmission data
        feuz_checkout_info.refresh_from_db()
        assert feuz_checkout_info.invoice_transmitted_at.date() == date(2024, 4, 21)
        abacus_transmission_list = feuz_checkout_info.additional_json_data.get(
            "abacus_transmission_list", []
        )
        assert len(abacus_transmission_list) == 2
        assert abacus_transmission_list[1]["transmitted_at"][0:10] == "2024-04-21"

        # check if files got created
        debitor_filepath = os.path.join(tmppath, "debitor/myVBV_debi_60000012.xml")
        assert os.path.exists(debitor_filepath)

        with open(debitor_filepath) as debitor_file:
            debi_content = debitor_file.read()
            assert "<CustomerNumber>60000012</CustomerNumber>" in debi_content
            assert "<Email>andreas.feuz@eiger-versicherungen.ch</Email>" in debi_content

        order_filepath = os.path.join(
            tmppath, "order/myVBV_orde_20240215083312_60000012_6000000124.xml"
        )
        assert os.path.exists(order_filepath)
        with open(order_filepath) as order_file:
            order_content = order_file.read()
            assert (
                "<ReferencePurchaseOrder>24021508331287484</ReferencePurchaseOrder>"
                in order_content
            )
            assert "<CustomerNumber>60000012</CustomerNumber>" in order_content