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

SFTP test should run sequentially -> migrate these tests to pytest

This commit is contained in:
Daniel Egger 2024-06-19 17:16:14 +02:00
parent ca01bf1196
commit d2685f85a8
10 changed files with 147 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -779,7 +779,7 @@ if APP_ENVIRONMENT == "local":
# django-extensions
# ------------------------------------------------------------------------------
# https://django-extensions.readthedocs.io/en/latest/installation_instructions.html#configuration
INSTALLED_APPS += ["django_extensions", "django_watchfiles"] # noqa F405
INSTALLED_APPS += ["django_extensions"] # noqa F405
else:
# not local
# SECURITY

29
server/conftest.py Normal file
View File

@ -0,0 +1,29 @@
import pytest
from _pytest.runner import runtestprotocol
@pytest.hookimpl(tryfirst=True)
def pytest_collection_modifyitems(config, items):
parallel_items = []
serial_items = []
for item in items:
if "serial" in item.keywords:
serial_items.append(item)
else:
parallel_items.append(item)
# Modify the collection to run serial tests first
config.serial_items = serial_items
items[:] = parallel_items
@pytest.hookimpl(tryfirst=True)
def pytest_sessionfinish(session, exitstatus):
config = session.config
if hasattr(config, "serial_items") and config.serial_items:
serial_items = config.serial_items
# Run serial tests one by one
for item in serial_items:
runtestprotocol(item, nextitem=None)

211
server/integration_tests/abacus_sftp/test_abacus_sftp.py
View File

@ -6,8 +6,8 @@ from io import StringIO
from subprocess import Popen
from time import sleep
import pytest
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
@ -17,116 +17,115 @@ 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()
@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 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,
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")
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 = 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_60000012_20240215083312_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
)
sleep(3)
assert "<CustomerNumber>60000012</CustomerNumber>" in order_content
feuz_checkout_info.refresh_from_db()
assert feuz_checkout_info.abacus_ssh_upload_done
class AbacusSftpServerTestCase(BaseAbacusSftpServerTestCase):
def test_canWriteFile_toFakeSftpServer(self):
with AbacusSftpClient() as client:
files = client.listdir(".")
self.assertSetEqual({"debitor", "order"}, set(files))
# calling `abacus_ssh_upload` a second time will not upload files again...
os.remove(debitor_filepath)
os.remove(order_filepath)
str_file = StringIO()
str_file.write("Hello world\n")
str_file.seek(0)
client.putfo(str_file, "hello.txt")
abacus_ssh_upload(feuz_checkout_info)
files = client.listdir(".")
self.assertSetEqual({"debitor", "order", "hello.txt"}, set(files))
debitor_filepath = os.path.join(tmppath, "debitor/myVBV_debi_60000012.xml")
assert not os.path.exists(debitor_filepath)
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))
order_filepath = os.path.join(
tmppath, "order/myVBV_orde_60000012_20240215083312_6000000124.xml"
)
assert not os.path.exists(order_filepath)

2
server/pytest.ini
View File

@ -2,3 +2,5 @@
addopts = --ds=config.settings.test --no-migrations
python_files = tests.py test_*.py
norecursedirs = node_modules
markers =
serial: marks tests as serial (not to be run in parallel)

4
server/requirements/requirements-dev.in
View File

@ -10,6 +10,7 @@ django-stubs # https://github.com/typeddjango/django-stubs
pytest # https://github.com/pytest-dev/pytest
pytest-sugar # https://github.com/Frozenball/pytest-sugar
pytest-xdist #
pytest-order
djangorestframework-stubs # https://github.com/typeddjango/djangorestframework-stubs
@ -33,9 +34,6 @@ django-coverage-plugin # https://github.com/nedbat/django_coverage_plugin
pytest-django # https://github.com/pytest-dev/pytest-django
freezegun # https://github.com/spulec/freezegun
# django-watchfiles custom PR
https://github.com/q0w/django-watchfiles/archive/issue-1.zip
# code checking
truffleHog

10
server/requirements/requirements-dev.txt
View File

@ -131,7 +131,6 @@ django==3.2.20
# django-stubs-ext
# django-taggit
# django-treebeard
# django-watchfiles
# djangorestframework
# drf-spectacular
# graphene-django
@ -186,8 +185,6 @@ django-taggit==4.0.0
# via wagtail
django-treebeard==4.7
# via wagtail
django-watchfiles @ https://github.com/q0w/django-watchfiles/archive/issue-1.zip
# via -r requirements-dev.in
djangorestframework==3.14.0
# via
# -r requirements.in
@ -416,10 +413,13 @@ pytest==7.4.0
# via
# -r requirements-dev.in
# pytest-django
# pytest-order
# pytest-sugar
# pytest-xdist
pytest-django==4.5.2
# via -r requirements-dev.in
pytest-order==1.2.1
# via -r requirements-dev.in
pytest-sugar==0.9.7
# via -r requirements-dev.in
pytest-xdist==3.5.0
@ -609,9 +609,7 @@ wagtail-headless-preview==0.6.0
wagtail-localize==1.5.1
# via -r requirements.in
watchfiles==0.19.0
# via
# django-watchfiles
# uvicorn
# via uvicorn
wcwidth==0.2.6
# via prompt-toolkit
webencodings==0.5.1

3
server/requirements/requirements.txt
View File

@ -31,8 +31,7 @@ azure-core==1.29.1
azure-identity==1.14.0
# via -r requirements.in
azure-storage-blob==12.17.0
# via
# -r requirements.in
# via -r requirements.in
bcrypt==4.0.1
# via paramiko
beautifulsoup4==4.11.2

2
server/run_tests.sh
View File

@ -3,4 +3,4 @@
cd "$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
# limit test to 6 parallel processes, otherwise ratelimit of s3 could be hit
pytest --numprocesses auto --maxprocesses=6 --junitxml=../test-reports/coverage.xml
pytest --numprocesses auto --maxprocesses=6 --dist=loadscope --junitxml=../test-reports/coverage.xml $1

2
server/run_tests_coverage.sh
View File

@ -3,7 +3,7 @@
set -e
cd "$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
coverage run -m pytest --numprocesses auto --maxprocesses=6 --junitxml=../test-reports/coverage.xml $1
pytest --numprocesses auto --maxprocesses=6 --dist=loadscope --junitxml=../test-reports/coverage.xml $1
coverage_python=`coverage report -m | tail -n1 | awk '{print $4}'`
commit=`git rev-parse HEAD`

4
server/vbv_lernwelt/shop/tests/test_abacus_invoice.py
View File

@ -46,7 +46,7 @@ class AbacusInvoiceTestCase(TestCase):
invoice_xml_filename, "myVBV_orde_60000012_20240215083312_6000000124.xml"
)
# print(invoice_xml_content)
print(invoice_xml_content)
assert "<CustomerNumber>60000012</CustomerNumber>" in invoice_xml_content
assert (
"<PurchaseOrderDate>2024-02-15</PurchaseOrderDate>" in invoice_xml_content
@ -61,7 +61,7 @@ class AbacusInvoiceTestCase(TestCase):
)
assert "<DeliveryDate>2024-02-15</DeliveryDate>" in invoice_xml_content
assert (
"<Text>Versicherungsvermittler/-in VBV - 2024-02-15 - Feuz Andreas</Text>"
"<Text>Versicherungsvermittler/-in VBV, 2024-02-15, Feuz Andreas</Text>"
in invoice_xml_content
)