ramon
/
vbv
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Test abacus sftp upload

This commit is contained in:
Daniel Egger 2024-05-31 17:15:45 +02:00
parent c89914107f
commit 516079ba10
8 changed files with 185 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/config/settings/test.py
View File

@ -23,6 +23,8 @@ EMAIL_BACKEND = "django.core.mail.backends.locmem.EmailBackend"
WHITENOISE_MANIFEST_STRICT = False
AWS_S3_FILE_OVERWRITE = True
ABACUS_EXPORT_SFTP_PORT = 34343
class DisableMigrations(dict):
def __contains__(self, item):

129
server/integration_tests/abacus_sftp/test_abacus_sftp.py Normal file
View File

@ -0,0 +1,129 @@
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.assertEqual(["debitor", "order"], files)
str_file = StringIO()
str_file.write("Hello world\n")
str_file.seek(0)
client.putfo(str_file, "hello.txt")
files = client.listdir(".")
self.assertEqual({"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 order file 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.assertTrue(os.path.exists(debitor_filepath))
order_filepath = (
self.tmppath + "/order/myVBV_orde_60000012_20240215083312_6000000124.xml"
)
self.assertFalse(os.path.exists(order_filepath))

80
server/vbv_lernwelt/shop/datatrans_fake_server.py
View File

@ -2,7 +2,9 @@ import hashlib
import hmac
import json
import threading
import time
import requests
from django.conf import settings
from django.http import HttpResponse, JsonResponse
from django.shortcuts import redirect
@ -15,80 +17,6 @@ from vbv_lernwelt.core.models import User
@csrf_exempt
@django_view_authentication_exempt
def fake_datatrans_api_view(request, api_url=""):
# if api_url == "/redirect":
# fake_tamedia_token = request.GET.get("token")
# pai = fake_tamedia_token.split(":")[1]
# sub = SubhubCustomerSubscription.objects.filter(id=pai).first()
#
# header = f"<h1>fake tamedia activation for {pai}"
#
# if not sub:
# return HttpResponse(
# content=f"""
# {header}
# <p>no subscription found</p>
# """,
# status=404,
# )
#
# if request.method == "GET":
# if (
# sub
# and sub.partner_status
# == SubhubCustomerSubscription.PARTNER_STATUS_ENROLLED
# ):
# return HttpResponse(
# content=f"""
# {header}
# <div>
# <input type="text" name="partnerSubscriptionId" value="{pai}">
# <form action="{create_register_url(fake_tamedia_token)}" method="post">
# <button type="submit" data-cy="register-button">Register</button>
# </form>
# </div>
# """,
# status=200,
# )
# else:
# return HttpResponse(
# content=f"""
# {header}
# <p>already activated</p>
# """,
# status=200,
# )
# if request.method == "POST":
# if sub:
# response = requests.post(
# f"{settings.APPLICATION_ABSOLUTE_URL}/subhub/ottwebhook",
# json={
# "PartnerIntegration": {
# "effectiveDate": datetime.now().isoformat(),
# "eventType": "activation",
# "offerId": sub.subscription_choice.partner_product_id,
# "optionalAttributes": None,
# "pai": pai,
# "partnerType": "Tamedia",
# "transactionId": str(uuid.uuid4()),
# },
# "eventId": str(uuid.uuid4()),
# "eventType": "OTT Partner Events",
# "publisherId": "Partner Events",
# "status": "new",
# "timestamp": datetime.now().isoformat(),
# },
# auth=HTTPBasicAuth(
# "swisscom_ott_webhook",
# "swisscom-ott-webhook-rLaYG0btVJMPtfnzfLilZtm50",
# ),
# )
# print(response)
# return redirect(f"{create_register_url(fake_tamedia_token)}")
#
# if api_url.startswith("/enroll") and request.method == "POST":
# return HttpResponse(status=204)
#
if api_url == "/v1/transactions" and request.method == "POST":
data = json.loads(request.body.decode("utf-8"))
user = User.objects.get(id=data["user_id"])
@ -107,10 +35,6 @@ def fake_datatrans_pay_view(request, api_url=""):
def call_transaction_complete_webhook(
webhook_url, transaction_id, datatrans_status="settled"
):
import time
import requests
time.sleep(1)
payload = {

26
server/vbv_lernwelt/shop/invoice/abacus.py
View File

@ -4,12 +4,10 @@ from xml.dom import minidom
from xml.etree.ElementTree import Element, SubElement, tostring
import structlog
from django.conf import settings
from paramiko.client import AutoAddPolicy, SSHClient
from vbv_lernwelt.shop.invoice.abacus_sftp_client import AbacusSftpClient
from vbv_lernwelt.shop.models import CheckoutInformation
logger = structlog.get_logger(__name__)
@ -22,7 +20,11 @@ def abacus_ssh_upload(checkout_information: CheckoutInformation):
abacus_ssh_upload_invoice(
customer_xml_filename, customer_xml_content, folder="debitor"
)
abacus_ssh_upload_invoice(invoice_xml_filename, invoice_xml_content, folder="order")
if not checkout_information.abacus_ssh_upload_done:
abacus_ssh_upload_invoice(invoice_xml_filename, invoice_xml_content, folder="order")
checkout_information.abacus_ssh_upload_done = True
checkout_information.save()
def create_invoice_xml(checkout_information: CheckoutInformation):
@ -210,20 +212,8 @@ def abacus_ssh_upload_invoice(
filename: str, content_xml: str, folder: str = ""
) -> None:
invoice_io = BytesIO(content_xml.encode("utf-8"))
ssh_client = SSHClient()
print(invoice_io)
try:
ssh_client.set_missing_host_key_policy(AutoAddPolicy())
ssh_client.connect(
hostname=settings.ABACUS_EXPORT_SFTP_HOST,
port=settings.ABACUS_EXPORT_SFTP_PORT,
username=settings.ABACUS_EXPORT_SFTP_USERNAME,
password=settings.ABACUS_EXPORT_SFTP_PASSWORD,
)
with ssh_client.open_sftp() as sftp_client:
with AbacusSftpClient() as sftp_client:
path = filename
if folder:
path = f"{folder}/{filename}"
@ -232,5 +222,3 @@ def abacus_ssh_upload_invoice(
except Exception as e:
logger.error("Could not upload invoice", exc_info=e)
finally:
ssh_client.close()

26
server/vbv_lernwelt/shop/invoice/abacus_sftp_client.py Normal file
View File

@ -0,0 +1,26 @@
from django.conf import settings
from paramiko.client import SSHClient, AutoAddPolicy
def _create_abacus_sftp_client():
ssh_client = SSHClient()
ssh_client.set_missing_host_key_policy(AutoAddPolicy())
ssh_client.connect(
hostname=settings.ABACUS_EXPORT_SFTP_HOST,
port=settings.ABACUS_EXPORT_SFTP_PORT,
username=settings.ABACUS_EXPORT_SFTP_USERNAME,
password=settings.ABACUS_EXPORT_SFTP_PASSWORD,
)
return ssh_client.open_sftp(), ssh_client
class AbacusSftpClient:
def __enter__(self):
(self.sftp_client, self.ssh_client) = _create_abacus_sftp_client()
return self.sftp_client
def __exit__(self, exc_type, exc_value, exc_traceback):
# self.sftp_client.close()
self.ssh_client.close()

18
server/vbv_lernwelt/shop/migrations/0014_checkoutinformation_abacus_ssh_upload_done.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 3.2.20 on 2024-05-31 15:17
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('shop', '0013_checkoutinformation_abacus_order_id'),
]
operations = [
migrations.AddField(
model_name='checkoutinformation',
name='abacus_ssh_upload_done',
field=models.BooleanField(default=False),
),
]

1
server/vbv_lernwelt/shop/models.py
View File

@ -101,6 +101,7 @@ class CheckoutInformation(models.Model):
# is only set by abacus invoice export code
abacus_order_id = models.BigIntegerField(unique=True, null=True, blank=True)
abacus_ssh_upload_done = models.BooleanField(default=False)
def set_increment_abacus_order_id(self):
if self.abacus_order_id:

41
server/vbv_lernwelt/shop/tests/test_invoice.py → server/vbv_lernwelt/shop/tests/test_abacus_invoice.py
View File

@ -1,5 +1,4 @@
from datetime import date, datetime
from unittest.mock import create_autospec
from django.test import TestCase
@ -7,14 +6,11 @@ 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 (
AbacusInvoiceCreator,
create_customer_xml,
create_invoice_xml,
render_customer_xml,
render_invoice_xml,
)
from vbv_lernwelt.shop.invoice.repositories import InvoiceRepository
from vbv_lernwelt.shop.models import CheckoutInformation
from vbv_lernwelt.shop.tests.factories import CheckoutInformationFactory
USER_USERNAME = "testuser"
@ -219,40 +215,3 @@ class AbacusInvoiceTestCase(TestCase):
self.maxDiff = None
self.assertXMLEqual(expected_xml, customer_xml)
def test_create_invoice_calls_upload(self):
# GIVEN
repository_mock = create_autospec(InvoiceRepository)
creator = AbacusInvoiceCreator(repository=repository_mock)
expected_filename = "test.xml"
checkout_information = CheckoutInformation.objects.create(
user=self.user,
transaction_id="12345",
product_sku="001",
product_name="Test Product",
product_description="Test Product Description",
product_price=1000,
state="initialized",
)
# WHEN
creator.create_invoice(
checkout_information=checkout_information,
filename=expected_filename,
)
# THEN
repository_mock.upload_invoice.assert_called_once()
uploaded_invoice, uploaded_filename = repository_mock.upload_invoice.call_args[
0
]
assert uploaded_filename == expected_filename
assert "<CustomerNumber>12345</CustomerNumber>" in uploaded_invoice
assert "<ItemNumber>1</ItemNumber>" in uploaded_invoice
assert "<ProductNumber>001</ProductNumber>" in uploaded_invoice
assert "<QuantityOrdered>1</QuantityOrdered>" in uploaded_invoice
assert "<Text>Test Product Description</Text>" in uploaded_invoice