majic-ansible-roles Changeset

Changeset - c322cfcb5e2d

Parent rev.

Child rev.

[Not reviewed]

0 1 1

Branko Majic (branko) - 2 months ago 2025-02-08 00:13:25
branko@majic.rs

MAR-242: Refactor file upload tests:

- Use fixtures for grabbing the server host and cleaning up the
uploads directory. Deduplicates the test code a bit as well.

2 files changed with 60 insertions and 41 deletions:

roles/xmpp_server/molecule/default/tests/conftest.py

roles/xmpp_server/molecule/default/tests/test_client.py

0 comments (0 inline, 0 general) First comment

roles/xmpp_server/molecule/default/tests/conftest.py

➞

Show inline comments

@@ new file 100644 @@
 import os
 import time
 import pytest
 import testinfra.utils.ansible_runner
 @pytest.fixture
 def server_host(host, server):
     """
     Helper fixture that returns (Ansible-backed) server host based
     on the passed-in server name and calling host's distribution
     release.
     For example, if the server name is 'paramters-mandatory', and the
     calling host distribution is 'bookworm', the fixture will look for
     server called 'parameters-mandatory-bookworm' in the Ansible
     inventory.
     """
     ansible_facts = host.ansible("setup")["ansible_facts"]
     ansible_distribution_release = ansible_facts['ansible_distribution_release']
     ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
     server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])
     return server_host
 @pytest.fixture
 def server_clean_domain_uploads(server_host, domain):
     """
     Helper fixture that cleans the XMPP (Prosody) server's HTTP
     file uploads directory for the specified domain. This will wipe
     both the files and the stored upload-related metadata.
     The XMPP server can be considered fully operational once the
     fixture finishes the clean-up.
     """
     with server_host.sudo():
         server_host.run("rm -rf %s/*", f"/var/lib/prosody/upload%2e{domain}/")
         server_host.run("systemctl restart prosody")
         # Wait for Prosody to become available. Time-out after one
         # second.
         attempts = 1
         while server_host.run("prosodyctl status").rc != 0:
             time.sleep(0.1)
             attempts += 1
             if attempts > 10:
                 raise Exception("Prosody failed to restart after file upload directory cleanup.")

roles/xmpp_server/molecule/default/tests/test_client.py

➞

Show inline comments

 import os
 import time
 import uuid
 import pytest
     ["jane.doe", "janepassword", "domain2", "parameters-optional"],
     ["mick.doe", "mickpassword", "domain3", "parameters-optional"],
 ])
 def test_http_file_upload(host, username, password, domain, server):
 @pytest.mark.usefixtures("server_clean_domain_uploads")
 def test_http_file_upload(host, server_host, username, password, domain):
     """
     Tests if http file upload works correctly.
     """
     upload_directory_path = f"/var/lib/prosody/upload%2e{domain}/http_file_share"
     # Prepare file for transfer.
     expected_content = str(uuid.uuid4())
     create_sample_file = host.run("echo -n %s > /tmp/http_file_upload_sample.txt", expected_content)
     assert create_sample_file.rc == 0
     # Path where uploaded file will end-up.
     upload_directory_path = f"/var/lib/prosody/upload%2e{domain}/http_file_share"
     # Find the host that serves the domain. Used for validating uploaded content.
     ansible_facts = host.ansible("setup")["ansible_facts"]
     ansible_distribution_release = ansible_facts['ansible_distribution_release']
     ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
     server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])
     # Clean up leftovers from previous run.
     with server_host.sudo():
         server_host.run("rm -rf %s/*", upload_directory_path)
     # Upload the file.
     send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
                     f"--http-upload /tmp/http_file_upload_sample.txt "
     ["jane.doe", "janepassword", "domain2", "parameters-optional"],
     ["mick.doe", "mickpassword", "domain3", "parameters-optional"],
 ])
 def test_http_file_share_size_limit(host, username, password, domain, server):
 @pytest.mark.usefixtures("server_clean_domain_uploads")
 def test_http_file_share_size_limit(host, username, password, domain):
     """
     Tests the maximum file size for files uploaded via XEP-0363.
     """
     file_size_limit = 10 * 1024 * 1024
     # Find the host that serves the domain. Used for validating uploaded content.
     ansible_facts = host.ansible("setup")["ansible_facts"]
     ansible_distribution_release = ansible_facts['ansible_distribution_release']
     ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
     server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])
     # Clean up leftovers from previous run.
     with server_host.sudo():
         server_host.run("rm -rf %s/*", f"/var/lib/prosody/upload%2e{domain}/")
         server_host.run("systemctl restart prosody")
         time.sleep(1)
     # Test exact size limit.
     create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=%sB count=1", str(file_size_limit))
     assert create_sample_file.rc == 0
     ["jane.doe", "janepassword", "domain2", "parameters-optional"],
     ["mick.doe", "mickpassword", "domain3", "parameters-optional"],
 ])
 def test_http_file_share_daily_quota(host, username, password, domain, server):
 @pytest.mark.usefixtures("server_clean_domain_uploads")
 def test_http_file_share_daily_quota(host, username, password, domain):
     """
     Tests the user's daily quota for files uploaded via XEP-0363.
     """
     file_size_limit = 10 * 1024 * 1024
     # Find the host that serves the domain. Used for validating uploaded content.
     ansible_facts = host.ansible("setup")["ansible_facts"]
     ansible_distribution_release = ansible_facts['ansible_distribution_release']
     ansible_runner = testinfra.utils.ansible_runner.AnsibleRunner(os.environ['MOLECULE_INVENTORY_FILE'])
     server_host = ansible_runner.get_host(ansible_runner.get_hosts(f'{server}-{ansible_distribution_release}')[0])
     # Clean up leftovers from previous run.
     with server_host.sudo():
         server_host.run("rm -rf %s/*", f"/var/lib/prosody/upload%2e{domain}/")
         server_host.run("systemctl restart prosody")
         time.sleep(1)
     # Fill-up the daily quota.
     create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=%sB count=1", str(file_size_limit))
     assert create_sample_file.rc == 0

0 comments (0 inline, 0 general) First comment