Files
@ 2274a6ecbb76
Branch filter:
Location: majic-ansible-roles/roles/xmpp_server/molecule/default/tests/test_client.py
2274a6ecbb76
8.7 KiB
text/x-python
MAR-242: Mark some of the HTTP file sharing tests as slow:
- Update develpment instructions to include information on how to run
individual tests etc.
- Update develpment instructions to include information on how to run
individual tests etc.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 | import os
import uuid
import pytest
import testinfra.utils.ansible_runner
testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(
os.environ['MOLECULE_INVENTORY_FILE']).get_hosts('clients')
@pytest.mark.parametrize('server', ['parameters-mandatory', 'parameters-optional'])
@pytest.mark.parametrize('port', [5222, 5223, 5000, 5269, 5281])
@pytest.mark.parametrize('ip_protocol', [4, 6])
def test_connectivity(host, server, port, ip_protocol):
"""
Tests connectivity to the XMPP server (ports that should be reachable).
"""
with host.sudo():
scan = host.run('nmap -%s -p %s -oG - %s', str(ip_protocol), str(port), server)
assert scan.rc == 0
assert "Ports: %d/open/tcp//" % port in scan.stdout
@pytest.mark.parametrize("username, password, domain", [
["john.doe", "johnpassword", "domain1"],
["jane.doe", "janepassword", "domain2"],
])
def test_tls(host, username, password, domain):
"""
Tests if TLS works as expected.
"""
send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
f"--username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"{username}@{domain}")
assert send.rc == 0
assert "<body>Hello</body>" in send.stderr
send = host.run(f"echo 'Hello' | go-sendxmpp --debug --tls "
f"--username {username}@{domain} --password {password} --jserver {domain}:5223 "
f"{username}@{domain}")
assert send.rc == 0
assert "<body>Hello</body>" in send.stderr
@pytest.mark.parametrize("username, password, domain", [
["john.doe", "johnpassword", "domain1"],
["jane.doe", "janepassword", "domain2"],
])
def test_authentication_requires_tls(host, username, password, domain):
"""
Tests if STARTTLS is required.
"""
send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
f"--username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"{username}@{domain}")
assert send.rc == 0
assert "<starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'><required/></starttls>" in send.stderr
@pytest.mark.parametrize("username, password, domain", [
["john.doe", "johnpassword", "domain1"],
["jane.doe", "janepassword", "domain2"],
["mick.doe", "mickpassword", "domain3"],
])
def test_authentication(host, username, password, domain):
"""
Tests if authentication works correctly.
"""
send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
f"--username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"{username}@{domain}")
assert send.rc == 0
send = host.run(f"echo 'Hello' | go-sendxmpp --debug --tls "
f"--username {username}@{domain} --password {password} --jserver {domain}:5223 "
f"{username}@{domain}")
assert send.rc == 0
@pytest.mark.parametrize("target_username, target_domain", [
["john.doe", "domain1"],
["jane.doe", "domain2"],
])
def test_unauthorized_users_rejected(host, target_username, target_domain):
"""
Tests if unauthorized users (present in LDAP, but not member of correct
group) are rejected from accessing the XMPP server.
"""
send = host.run(f"echo 'Hello' | go-sendxmpp --debug "
f"--username noxmpp@{target_domain} --password noxmpppassword --jserver {target_domain}:5222 "
f"{target_username}@{target_domain}")
assert send.rc != 0
assert "Unable to authorize you with the authentication credentials you've sent" in send.stderr
@pytest.mark.parametrize("username, password, domain, server", [
["john.doe", "johnpassword", "domain1", "parameters-mandatory"],
["jane.doe", "janepassword", "domain2", "parameters-optional"],
["mick.doe", "mickpassword", "domain3", "parameters-optional"],
])
@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
# 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 "
f"{username}@{domain}")
assert send.rc == 0
assert "No http upload component found." not in send.stderr
# Verify content on server.
with server_host.sudo():
upload_directory = server_host.file(upload_directory_path)
assert upload_directory.is_directory
assert upload_directory.user == "prosody"
assert upload_directory.group == "prosody"
assert upload_directory.mode == 0o750
assert len(upload_directory.listdir()) == 1
uploaded_file_name = upload_directory.listdir()[0]
uploaded_file = server_host.file(os.path.join(upload_directory_path, uploaded_file_name))
assert uploaded_file.is_file
assert uploaded_file.user == "prosody"
assert uploaded_file.group == "prosody"
assert uploaded_file.mode == 0o640
assert uploaded_file.content_string == expected_content
@pytest.mark.parametrize("username, password, domain, server, file_size_limit", [
["john.doe", "johnpassword", "domain1", "parameters-mandatory", 10 * 1024 * 1024],
["jane.doe", "janepassword", "domain2", "parameters-optional", 20 * 1024 * 1024],
["mick.doe", "mickpassword", "domain3", "parameters-optional", 20 * 1024 * 1024],
])
@pytest.mark.usefixtures("server_clean_domain_uploads")
@pytest.mark.slow
def test_http_file_share_size_limit(host, username, password, domain, file_size_limit):
"""
Tests the maximum file size for files uploaded via XEP-0363.
"""
# 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
send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"--http-upload /tmp/http_file_upload_sample.txt "
f"{username}@{domain}")
assert "file-too-large" not in send.stderr
# Test exceeded 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 + 1))
assert create_sample_file.rc == 0
send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"--http-upload /tmp/http_file_upload_sample.txt "
f"{username}@{domain}")
assert "file-too-large" in send.stderr
@pytest.mark.parametrize("username, password, domain, server, file_size_limit, user_daily_quota", [
["john.doe", "johnpassword", "domain1", "parameters-mandatory", 10 * 1024 * 1024, 100 * 1024 * 1024],
["jane.doe", "janepassword", "domain2", "parameters-optional", 20 * 1024 * 1024, 70 * 1024 * 1024],
["mick.doe", "mickpassword", "domain3", "parameters-optional", 20 * 1024 * 1024, 70 * 1024 * 1024],
])
@pytest.mark.usefixtures("server_clean_domain_uploads")
@pytest.mark.slow
def test_http_file_share_daily_quota(host, username, password, domain, file_size_limit, user_daily_quota):
"""
Tests the user's daily quota for files uploaded via XEP-0363.
"""
remaining_quota = user_daily_quota
while remaining_quota > 0:
file_size = file_size_limit if file_size_limit < remaining_quota else remaining_quota
create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=%sB count=1", str(file_size))
assert create_sample_file.rc == 0
send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"--http-upload /tmp/http_file_upload_sample.txt "
f"{username}@{domain}")
assert send.rc == 0
remaining_quota -= file_size
# Test exceeded daily quota.
create_sample_file = host.run("dd if=/dev/zero of=/tmp/http_file_upload_sample.txt bs=1B count=1")
assert create_sample_file.rc == 0
send = host.run(f"go-sendxmpp --debug --username {username}@{domain} --password {password} --jserver {domain}:5222 "
f"--http-upload /tmp/http_file_upload_sample.txt "
f"{username}@{domain}")
assert "Daily quota reached" in send.stderr
|