133 lines
4.7 KiB
Python
133 lines
4.7 KiB
Python
import os
|
|
import shutil
|
|
import tempfile
|
|
from datetime import datetime
|
|
from io import StringIO
|
|
from subprocess import Popen
|
|
from time import sleep
|
|
|
|
from django.conf import settings
|
|
from django.test import TransactionTestCase
|
|
|
|
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
|
|
|
|
|
|
class BaseAbacusSftpServerTestCase(TransactionTestCase):
|
|
def setUp(self):
|
|
self.start_sftp_server()
|
|
|
|
def tearDown(self):
|
|
if self.sftp_server:
|
|
self.sftp_server.kill()
|
|
|
|
def start_sftp_server(self):
|
|
self.tmppath = tempfile.mkdtemp()
|
|
print(self.tmppath)
|
|
shutil.rmtree(self.tmppath)
|
|
os.mkdir(self.tmppath)
|
|
os.mkdir(self.tmppath + "/debitor")
|
|
os.mkdir(self.tmppath + "/order")
|
|
self.sftp_server = Popen(
|
|
f"sftpserver -p {settings.ABACUS_EXPORT_SFTP_PORT} -l INFO",
|
|
shell=True,
|
|
cwd=self.tmppath,
|
|
)
|
|
sleep(3)
|
|
|
|
|
|
class AbacusSftpServerTestCase(BaseAbacusSftpServerTestCase):
|
|
def test_canWriteFile_toFakeSftpServer(self):
|
|
with AbacusSftpClient() as client:
|
|
files = client.listdir(".")
|
|
self.assertSetEqual({"debitor", "order"}, set(files))
|
|
|
|
str_file = StringIO()
|
|
str_file.write("Hello world\n")
|
|
str_file.seek(0)
|
|
client.putfo(str_file, "hello.txt")
|
|
|
|
files = client.listdir(".")
|
|
self.assertSetEqual({"debitor", "order", "hello.txt"}, set(files))
|
|
|
|
|
|
class AbacusInvoiceUploadTestCase(BaseAbacusSftpServerTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
add_countries(small_set=True)
|
|
create_default_users()
|
|
|
|
def test_upload_abacus_xml(self):
|
|
# 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")
|
|
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",
|
|
)
|
|
feuz_checkout_info.created_at = datetime(2024, 2, 15, 8, 33, 12, 0)
|
|
|
|
abacus_ssh_upload(feuz_checkout_info)
|
|
|
|
# check if files got created
|
|
debitor_filepath = self.tmppath + "/debitor/myVBV_debi_60000012.xml"
|
|
self.assertTrue(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 = (
|
|
self.tmppath + "/order/myVBV_orde_60000012_20240215083312_6000000124.xml"
|
|
)
|
|
self.assertTrue(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()
|
|
self.assertTrue(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 = self.tmppath + "/debitor/myVBV_debi_60000012.xml"
|
|
self.assertFalse(os.path.exists(debitor_filepath))
|
|
|
|
order_filepath = (
|
|
self.tmppath + "/order/myVBV_orde_60000012_20240215083312_6000000124.xml"
|
|
)
|
|
self.assertFalse(os.path.exists(order_filepath))
|