Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • andersharrisson/csentry
  • ics-infrastructure/csentry
2 results
Show changes
Showing
with 1236 additions and 190 deletions
Implementation
==============
The main modules are described below.
Full source code is available in `GitLab <https://gitlab.esss.lu.se/ics-infrastructure/csentry>`_.
.. toctree::
:maxdepth: 2
models
commands
decorators
search
tasks
user_view
inventory_view
network_view
.. automodule:: app.inventory.views
:members:
:undoc-members:
.. automodule:: app.models
:members:
:undoc-members:
.. automodule:: app.network.views
:members:
:undoc-members:
.. automodule:: app.search
:members:
:undoc-members:
.. automodule:: app.tasks
:members:
:undoc-members:
.. automodule:: app.user.views
:members:
:undoc-members:
......@@ -6,7 +6,22 @@
Control System Entry
====================
Release |version|.
.. image:: https://img.shields.io/badge/code%20style-black-000000.svg
:target: https://github.com/ambv/black
.. image:: https://sonarqube.esss.lu.se/api/project_badges/measure?project=csentry&metric=alert_status
:target: https://sonarqube.esss.lu.se/dashboard?id=csentry
.. image:: https://sonarqube.esss.lu.se/api/project_badges/measure?project=csentry&metric=ncloc
:target: https://sonarqube.esss.lu.se/dashboard?id=csentry
.. image:: https://gitlab.esss.lu.se/ics-infrastructure/csentry/badges/master/pipeline.svg
:target: https://gitlab.esss.lu.se/ics-infrastructure/csentry
.. image:: https://gitlab.esss.lu.se/ics-infrastructure/csentry/badges/master/coverage.svg
:target: https://gitlab.esss.lu.se/ics-infrastructure/csentry
Release |release|.
Control System Entry is a web application that facilitates the tracking of physical items
and network devices.
......@@ -30,6 +45,7 @@ Please use the navigation sidebar on the left to begin.
.. toctree::
:hidden:
:caption: User documentation
:maxdepth: 3
inventory
......@@ -38,3 +54,13 @@ Please use the navigation sidebar on the left to begin.
profile
api
changelog
.. toctree::
:hidden:
:caption: Developer documentation
:maxdepth: 3
development/design
development/deployment
development/endpoints
development/implementation/index
"""Add interface description field
Revision ID: 5a2ca42797d9
Revises: 91b0093a5e13
Create Date: 2020-06-01 12:49:27.606851
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "5a2ca42797d9"
down_revision = "91b0093a5e13"
branch_labels = None
depends_on = None
def upgrade():
op.add_column("interface", sa.Column("description", sa.Text(), nullable=True))
def downgrade():
op.drop_column("interface", "description")
sphinx
sphinx_rtd_theme
sphinxcontrib-httpdomain
coverage
factory_boy
......
......@@ -19,11 +19,13 @@ python-dateutil
pyyaml
qrcode
whitenoise
ansible-tower-cli
ansible-tower-cli<3.3.9
rq
rq-dashboard
sentry-sdk
sqlalchemy<1.3
sqlalchemy-citext
sqlalchemy-continuum
openpyxl
uwsgi
WTForms>=2.1,<2.2
alembic==1.0.5
ansible-tower-cli==3.3.0
arrow==0.12.1
alembic==1.4.3
ansible-tower-cli==3.3.8
arrow==0.17.0
blinker==1.4
certifi==2018.11.29
cachelib==0.1.1
certifi==2020.6.20
chardet==3.0.4
Click==7.0
colorama==0.4.1
elasticsearch==7.0.4
click==7.1.2
colorama==0.4.4
elasticsearch==7.9.1
et-xmlfile==1.0.1
Flask==1.0.2
Flask-Admin==1.5.2
Flask-Caching==1.4.0
Flask-DebugToolbar==0.10.1
Flask-JWT-Extended==3.13.1
Flask==1.1.2
Flask-Admin==1.5.6
Flask-Caching==1.9.0
Flask-DebugToolbar==0.11.0
Flask-JWT-Extended==3.24.1
flask-ldap3-login==0.9.16
Flask-Login==0.4.1
Flask-Login==0.5.0
Flask-Mail==0.9.1
Flask-Migrate==2.3.1
Flask-Redis==0.3.0
Flask-Session==0.3.1
Flask-SQLAlchemy==2.3.2
Flask-WTF==0.14.2
idna==2.7
Flask-Migrate==2.5.3
flask-redis==0.4.0
Flask-Session==0.3.2
Flask-SQLAlchemy==2.4.4
Flask-WTF==0.14.3
idna==2.10
itsdangerous==1.1.0
jdcal==1.4
Jinja2==2.10
ldap3==2.5.1
Mako==1.0.7
MarkupSafe==1.1.0
openpyxl==2.5.12
Pillow==5.3.0
psycopg2==2.7.6.1
pyasn1==0.4.4
PyJWT==1.6.4
python-dateutil==2.7.5
python-editor==1.0.3
PyYAML==3.13
qrcode==6.0
redis==2.10.6
requests==2.20.1
rq==0.12.0
rq-dashboard==0.3.12
sentry-sdk==0.7.10
six==1.11.0
SQLAlchemy==1.2.14
sqlalchemy-citext==1.3.post0
SQLAlchemy-Continuum==1.3.6
SQLAlchemy-Utils==0.33.8
urllib3==1.24.1
uwsgi==2.0.18
Werkzeug==0.14.1
whitenoise==4.1.2
jdcal==1.4.1
Jinja2==2.11.2
ldap3==2.8.1
Mako==1.1.3
MarkupSafe==1.1.1
openpyxl==3.0.5
Pillow==8.0.0
psycopg2==2.8.6
pyasn1==0.4.8
PyJWT==1.7.1
python-dateutil==2.8.1
python-editor==1.0.4
PyYAML==5.3.1
qrcode==6.1
redis==3.5.3
requests==2.24.0
rq==1.5.2
rq-dashboard==0.6.1
sentry-sdk==0.19.1
six==1.15.0
SQLAlchemy==1.2.19
sqlalchemy-citext==1.7.0
SQLAlchemy-Continuum==1.3.11
SQLAlchemy-Utils==0.36.8
urllib3==1.25.11
uWSGI==2.0.19.1
Werkzeug==1.0.1
whitenoise==5.2.0
WTForms==2.1
......@@ -9,10 +9,12 @@ Pytest fixtures common to all functional tests.
:license: BSD 2-Clause, see LICENSE for more details.
"""
import redis
import pytest
import sqlalchemy as sa
from pytest_factoryboy import register
from flask_ldap3_login import AuthenticationResponse, AuthenticationResponseStatus
from rq import push_connection, pop_connection
from app.factory import create_app
from app.extensions import db as _db
from app.models import SearchableMixin, Host, Item, AnsibleGroup
......@@ -45,12 +47,14 @@ def app(request):
"TESTING": True,
"WTF_CSRF_ENABLED": False,
"SQLALCHEMY_DATABASE_URI": "postgresql://ics:icspwd@postgres/csentry_db_test",
"RQ_REDIS_URL": "redis://redis:6379/4",
"ELASTICSEARCH_INDEX_SUFFIX": "-test",
"ELASTICSEARCH_REFRESH": "true",
"CACHE_TYPE": "null",
"CACHE_NO_NULL_WARNING": True,
"CSENTRY_LDAP_GROUPS": {
"admin": ["CSEntry Admin"],
"auditor": ["CSEntry Auditor"],
"inventory": ["CSEntry User"],
},
"CSENTRY_NETWORK_SCOPES_LDAP_GROUPS": {
......@@ -134,8 +138,17 @@ def session(db, request):
Host.create_index()
AnsibleGroup.create_index()
# Setup RQ redis connection
redis_connection = redis.from_url(db.app.config["RQ_REDIS_URL"])
redis_connection.flushdb()
push_connection(redis_connection)
yield session
# Clean RQ redis connnection
redis_connection.flushdb()
pop_connection()
# ELASTICSEARCH_INDEX_SUFFIX is set to "-test"
# Delete all "*-test" indices after each test
db.app.elasticsearch.indices.delete("*-test", ignore=404)
......@@ -160,6 +173,21 @@ def patch_ldap_authenticate(monkeypatch):
response.status = AuthenticationResponseStatus.success
response.user_info = {"cn": "Admin User", "mail": "admin@example.com"}
response.user_groups = [{"cn": "CSEntry Admin"}]
elif username == "audit" and password == "auditpasswd":
response.status = AuthenticationResponseStatus.success
response.user_dn = "uid=audit,ou=Service accounts,dc=esss,dc=lu,dc=se"
response.user_info = {
"uid": ["audit"],
"cn": [],
"mail": [],
"dn": "uid=audit,ou=Service accounts,dc=esss,dc=lu,dc=se",
}
response.user_groups = [
{
"cn": ["CSEntry Auditor"],
"dn": "cn=CSEntry Auditor,ou=ICS,ou=Groups,dc=esss,dc=lu,dc=se",
}
]
elif username == "user_rw" and password == "userrw":
response.status = AuthenticationResponseStatus.success
response.user_info = {"cn": "User RW", "mail": "user_rw@example.com"}
......
......@@ -93,6 +93,7 @@ class ItemCommentFactory(factory.alchemy.SQLAlchemyModelFactory):
body = factory.Sequence(lambda n: f"comment{n}")
user = factory.SubFactory(UserFactory)
item = factory.SubFactory(ItemFactory)
class DomainFactory(factory.alchemy.SQLAlchemyModelFactory):
......
This diff is collapsed.
......@@ -31,6 +31,13 @@ def test_user_is_admin(user_factory):
assert user.is_admin
def test_user_is_auditor(user_factory):
user = user_factory(groups=["foo", "CSEntry User"])
assert not user.is_auditor
user = user_factory(groups=["foo", "CSEntry Auditor"])
assert user.is_auditor
def test_user_is_member_of_one_group(user_factory):
user = user_factory(groups=["one", "two"])
assert not user.is_member_of_one_group(["network", "admin"])
......@@ -42,6 +49,9 @@ def test_user_is_member_of_one_group(user_factory):
assert not user.is_member_of_one_group(["network"])
assert user.is_member_of_one_group(["network", "admin"])
assert user.is_member_of_one_group(["admin"])
user = user_factory(groups=["CSEntry Auditor"])
assert not user.is_member_of_one_group(["network", "admin"])
assert user.is_member_of_one_group(["auditor"])
def test_user_network_scopes(user_factory):
......@@ -51,6 +61,86 @@ def test_user_network_scopes(user_factory):
assert user.csentry_network_scopes == ["LabNetworks"]
@pytest.mark.parametrize(
"groups, sensitive_filter",
[
([], "sensitive:false"),
(["foo"], "sensitive:false"),
(
["CSEntry Lab"],
"sensitive:false OR (sensitive:true AND (scope:LabNetworks))",
),
(
["CSEntry Prod", "CSEntry User"],
"sensitive:false OR (sensitive:true AND (scope:ProdNetworks OR scope:FooNetworks))",
),
],
)
def test_user_sensitive_filter(user_factory, groups, sensitive_filter):
user = user_factory(groups=groups)
assert user.sensitive_filter == sensitive_filter
@pytest.mark.parametrize(
"scope_name, groups, sensitive, expected",
[
("ProdNetworks", ["CSEntry Admin"], False, True),
("ProdNetworks", ["CSEntry Admin"], True, True),
("ProdNetworks", ["CSEntry Prod"], False, True),
("ProdNetworks", ["CSEntry Prod"], True, True),
("ProdNetworks", ["CSEntry Lab"], False, True),
("ProdNetworks", ["CSEntry Lab"], True, False),
],
)
@pytest.mark.parametrize("admin_only", [False, True])
def test_user_can_view_host(
user_factory,
network_scope_factory,
network_factory,
interface_factory,
host_factory,
scope_name,
groups,
sensitive,
expected,
admin_only,
):
scope = network_scope_factory(name=scope_name)
network = network_factory(scope=scope, admin_only=admin_only, sensitive=sensitive)
host = host_factory()
interface_factory(name=host.name, host=host, network=network)
user = user_factory(groups=groups)
assert user.can_view_host(host) == expected
@pytest.mark.parametrize(
"scope_name, groups, sensitive, expected",
[
("ProdNetworks", ["CSEntry Admin"], False, True),
("ProdNetworks", ["CSEntry Admin"], True, True),
("ProdNetworks", ["CSEntry Prod"], False, True),
("ProdNetworks", ["CSEntry Prod"], True, True),
("ProdNetworks", ["CSEntry Lab"], False, True),
("ProdNetworks", ["CSEntry Lab"], True, False),
],
)
@pytest.mark.parametrize("admin_only", [False, True])
def test_user_can_view_network(
user_factory,
network_scope_factory,
network_factory,
scope_name,
groups,
sensitive,
expected,
admin_only,
):
scope = network_scope_factory(name=scope_name)
network = network_factory(scope=scope, admin_only=admin_only, sensitive=sensitive)
user = user_factory(groups=groups)
assert user.can_view_network(network) == expected
def test_user_has_access_to_network(
user_factory, network_scope_factory, network_factory
):
......@@ -404,13 +494,12 @@ def test_ansible_groups_children(ansible_group_factory, host_factory):
group1 = ansible_group_factory()
group2 = ansible_group_factory()
group3 = ansible_group_factory()
group1.children.append(group2)
group1.children.append(group3)
assert group1.children == [group2, group3]
group1.children = [group2, group3]
assert group1.children == sorted([group2, group3], key=lambda grp: grp.name)
assert group2.parents == [group1]
assert group3.parents == [group1]
group4 = ansible_group_factory(parents=[group1])
assert group1.children == [group2, group3, group4]
assert group1.children == sorted([group2, group3, group4], key=lambda grp: grp.name)
def test_ansible_groups_children_all_forbidden(ansible_group_factory):
......@@ -418,7 +507,7 @@ def test_ansible_groups_children_all_forbidden(ansible_group_factory):
group1 = ansible_group_factory()
group2 = ansible_group_factory()
with pytest.raises(ValidationError) as excinfo:
group1.children.append(all)
group1.children = [all]
assert (
f"Adding group 'all' as child to '{group1.name}' creates a recursive dependency loop"
in str(excinfo.value)
......@@ -436,7 +525,7 @@ def test_ansible_groups_no_recursive_dependency(ansible_group_factory):
group2 = ansible_group_factory(children=[group3])
group1 = ansible_group_factory(children=[group2])
with pytest.raises(ValidationError) as excinfo:
group3.children.append(group1)
group3.children = [group1]
assert (
f"Adding group '{group1.name}' as child to '{group3.name}' creates a recursive dependency loop"
in str(excinfo.value)
......@@ -446,10 +535,84 @@ def test_ansible_groups_no_recursive_dependency(ansible_group_factory):
def test_ansible_groups_no_child_of_itself(ansible_group_factory):
group1 = ansible_group_factory()
with pytest.raises(ValidationError) as excinfo:
group1.children.append(group1)
group1.children = [group1]
assert f"Group '{group1.name}' can't be a child of itself" in str(excinfo.value)
@pytest.mark.parametrize(
"grp_type",
[
models.AnsibleGroupType.STATIC,
models.AnsibleGroupType.DEVICE_TYPE,
models.AnsibleGroupType.IOC,
models.AnsibleGroupType.HOSTNAME,
],
)
def test_ansible_group_network_scope_children(ansible_group_factory, grp_type):
group = ansible_group_factory(type=grp_type)
scope_group = ansible_group_factory(
type=models.AnsibleGroupType.NETWORK_SCOPE, children=[group]
)
assert scope_group.children == [group]
assert group.parents == [scope_group]
@pytest.mark.parametrize(
"grp_type",
[
models.AnsibleGroupType.NETWORK,
models.AnsibleGroupType.NETWORK_SCOPE,
],
)
def test_ansible_group_network_scope_children_forbidden(
ansible_group_factory, grp_type
):
child_group = ansible_group_factory(name="mygroup", type=grp_type)
with pytest.raises(ValidationError) as excinfo:
ansible_group_factory(
type=models.AnsibleGroupType.NETWORK_SCOPE, children=[child_group]
)
assert (
f"can't set {str(grp_type).lower()} group 'mygroup' as a network scope child"
in str(excinfo.value)
)
@pytest.mark.parametrize(
"grp_type",
[
models.AnsibleGroupType.STATIC,
models.AnsibleGroupType.DEVICE_TYPE,
models.AnsibleGroupType.IOC,
models.AnsibleGroupType.HOSTNAME,
],
)
def test_ansible_group_network_parent(ansible_group_factory, grp_type):
group = ansible_group_factory(type=grp_type)
network_group = ansible_group_factory(
type=models.AnsibleGroupType.NETWORK, parents=[group]
)
assert network_group.parents == [group]
assert group.children == [network_group]
@pytest.mark.parametrize(
"grp_type",
[
models.AnsibleGroupType.NETWORK,
models.AnsibleGroupType.NETWORK_SCOPE,
],
)
def test_ansible_group_network_parent_forbidden(ansible_group_factory, grp_type):
group = ansible_group_factory(name="mygroup", type=grp_type)
with pytest.raises(ValidationError) as excinfo:
ansible_group_factory(type=models.AnsibleGroupType.NETWORK, parents=[group])
assert (
f"can't set {str(grp_type).lower()} group 'mygroup' as a network parent"
in str(excinfo.value)
)
def test_host_model(model_factory, item_factory, host_factory):
host1 = host_factory()
model1 = model_factory(name="EX3400")
......@@ -505,7 +668,7 @@ def test_ansible_dynamic_network_group(
assert group3.hosts == []
def test_ansible_dynamic_network_scope_group(
def test_ansible_dynamic_network_scope_group_hosts(
ansible_group_factory,
network_scope_factory,
network_factory,
......@@ -540,6 +703,68 @@ def test_ansible_dynamic_network_scope_group(
assert group3.hosts == []
@pytest.mark.parametrize(
"networks, groups, expected_names",
[
(
["network1", "network2", "network3"],
[],
[],
),
(
["network1", "network2", "network3"],
[("network2", models.AnsibleGroupType.NETWORK)],
["network2"],
),
(
["network1", "network2", "network3"],
[
("network2", models.AnsibleGroupType.NETWORK),
("network3", models.AnsibleGroupType.NETWORK),
],
["network2", "network3"],
),
(
["network1"],
[
("network2", models.AnsibleGroupType.NETWORK),
],
[],
),
(
["network1", "network2"],
[
("mygroup1", models.AnsibleGroupType.DEVICE_TYPE),
("network2", models.AnsibleGroupType.NETWORK),
("mygroup2", models.AnsibleGroupType.STATIC),
],
["mygroup1", "mygroup2", "network2"],
),
],
)
def test_ansible_dynamic_network_scope_group_children(
ansible_group_factory,
network_scope_factory,
network_factory,
networks,
groups,
expected_names,
):
name = "myscope"
scope = network_scope_factory(name=name)
group = ansible_group_factory(name=name, type=models.AnsibleGroupType.NETWORK_SCOPE)
for network in networks:
network_factory(vlan_name=network, scope=scope)
for grp, grp_type in groups:
ag = ansible_group_factory(name=grp, type=grp_type)
if grp_type != models.AnsibleGroupType.NETWORK:
ag.parents = [group]
expected = [
grp for grp in models.AnsibleGroup.query.all() if grp.name in expected_names
]
assert group.children == sorted(expected, key=lambda grp: grp.name)
def test_ansible_dynamic_ioc_group(ansible_group_factory, host_factory):
host1 = host_factory(name="host1", is_ioc=True)
host2 = host_factory(name="host2", is_ioc=True)
......@@ -800,6 +1025,21 @@ def test_host_items_sorted_with_mixed_stack_member(host_factory, item_factory):
]
def test_host_no_scope(host_factory):
host = host_factory()
assert host.scope is None
def test_host_scope(
host_factory, network_scope_factory, network_factory, interface_factory
):
scope = network_scope_factory()
network = network_factory(scope=scope)
host = host_factory()
interface_factory(name=host.name, host=host, network=network)
assert host.scope == scope
def test_cname_existing_host(db, host_factory, cname_factory):
host_factory(name="myhost")
with pytest.raises(ValidationError) as excinfo:
......@@ -845,36 +1085,36 @@ def test_task_awx_job_url(db, task_factory):
assert task5.awx_job_url == "https://awx.example.org/#/jobs/inventory/12"
@pytest.mark.parametrize("length", (1, 21, 50))
@pytest.mark.parametrize("length", (1, 25, 50))
def test_hostname_invalid_length(db, host_factory, length):
with pytest.raises(ValidationError) as excinfo:
host_factory(name="x" * length)
assert r"Host name shall match [a-z0-9\-]{2,20}" in str(excinfo.value)
assert r"Host name shall match ^[a-z0-9\-]{2,24}" in str(excinfo.value)
@pytest.mark.parametrize("name", ("my_host", "host@", "foo:bar", "U02.K02"))
def test_hostname_invalid_characters(db, host_factory, name):
with pytest.raises(ValidationError) as excinfo:
host_factory(name=name)
assert r"Host name shall match [a-z0-9\-]{2,20}" in str(excinfo.value)
assert r"Host name shall match ^[a-z0-9\-]{2,24}" in str(excinfo.value)
@pytest.mark.parametrize("length", (1, 26, 50))
@pytest.mark.parametrize("length", (1, 30, 50))
def test_interface_name_invalid_length(db, interface_factory, length):
with pytest.raises(ValidationError) as excinfo:
interface_factory(name="x" * length)
assert r"Interface name shall match [a-z0-9\-]{2,25}" in str(excinfo.value)
assert r"Interface name shall match ^[a-z0-9\-]{2,29}" in str(excinfo.value)
def test_interface_name_length(db, host_factory, interface_factory):
hostname = "x" * 20
hostname = "x" * 24
interface_name = hostname + "-yyyy"
host1 = host_factory(name=hostname)
interface_factory(name=interface_name, host=host1)
assert host1.interfaces[0].name == interface_name
with pytest.raises(ValidationError) as excinfo:
interface_factory(name=interface_name + "y", host=host1)
assert r"Interface name shall match [a-z0-9\-]{2,25}" in str(excinfo.value)
assert r"Interface name shall match ^[a-z0-9\-]{2,29}" in str(excinfo.value)
@pytest.mark.parametrize("ics_id", ("123", "AA123", "AAA1234"))
......@@ -950,6 +1190,55 @@ def test_network_scope_overlapping(address, network_scope_factory):
)
def test_network_scope_supernet_validation(network_scope_factory, network_factory):
scope = network_scope_factory(
first_vlan=3800, last_vlan=4000, supernet="172.30.0.0/16"
)
network1 = network_factory(
vlan_id=3800,
address="172.30.0.0/23",
first_ip="172.30.0.3",
last_ip="172.30.1.240",
scope=scope,
)
address = "192.168.0.0/16"
with pytest.raises(ValidationError) as excinfo:
scope.supernet = "192.168.0.0/16"
assert f"{network1.network_ip} is not a subnet of {address}" in str(excinfo.value)
def test_network_scope_first_vlan_validation(network_scope_factory, network_factory):
scope = network_scope_factory(
first_vlan=200, last_vlan=400, supernet="172.30.0.0/16"
)
network1 = network_factory(
vlan_id=220,
address="172.30.0.0/23",
first_ip="172.30.0.3",
last_ip="172.30.1.240",
scope=scope,
)
with pytest.raises(ValidationError) as excinfo:
scope.first_vlan = 230
assert f"First vlan shall be lower than {network1} vlan: 220" in str(excinfo.value)
def test_network_scope_last_vlan_validation(network_scope_factory, network_factory):
scope = network_scope_factory(
first_vlan=200, last_vlan=400, supernet="172.30.0.0/16"
)
network1 = network_factory(
vlan_id=220,
address="172.30.0.0/23",
first_ip="172.30.0.3",
last_ip="172.30.1.240",
scope=scope,
)
with pytest.raises(ValidationError) as excinfo:
scope.last_vlan = 210
assert f"Last vlan shall be greater than {network1} vlan: 220" in str(excinfo.value)
def test_host_sensitive_field_update_on_network_change(
network_scope_factory, network_factory, interface_factory, host_factory
):
......@@ -975,3 +1264,61 @@ def test_host_sensitive_field_update_on_network_change(
instances, nb = models.Host.search("sensitive:true")
assert nb == 1
assert instances[0].name == name
@pytest.mark.parametrize(
"dn,username,user_info,user_groups,expected_display_name,expected_email,expected_groups",
[
(
"uid=johndoe,ou=Users,dc=esss,dc=lu,dc=se",
"johndoe",
{"mail": "john.doe@example.org", "cn": "John Doe"},
[{"cn": "group2"}, {"cn": "group1"}],
"John Doe",
"john.doe@example.org",
["group1", "group2"],
),
(
"uid=johndoe,ou=Users,dc=esss,dc=lu,dc=se",
"johndoe",
{"mail": ["john.doe@example.org"], "cn": ["John Doe"]},
[{"cn": ["group2"]}, {"cn": ["group1"]}],
"John Doe",
"john.doe@example.org",
["group1", "group2"],
),
(
"uid=auditor,ou=Service accounts,dc=esss,dc=lu,dc=se",
"auditor",
{
"uid": ["auditor"],
"cn": [],
"mail": [],
"dn": "uid=csentry_svc,ou=Service accounts,dc=esss,dc=lu,dc=se",
},
[
{
"cn": ["csentry_auditors"],
"dn": "cn=csentry_auditors,ou=ICS,ou=Groups,dc=esss,dc=lu,dc=se",
}
],
"auditor",
"",
["csentry_auditors"],
),
],
)
def test_save_user(
dn,
username,
user_info,
user_groups,
expected_display_name,
expected_email,
expected_groups,
):
user = models.save_user(dn, username, user_info, user_groups)
assert user.username == username
assert user.display_name == expected_display_name
assert user.email == expected_email
assert user.groups == expected_groups
import pytest
@pytest.mark.parametrize(
"name, func, input_kwargs, output_args",
[
("my task1", "my_func1", {}, ""),
(
"my task2",
"my_func2",
{"arg1": "foo", "arg2": True},
"arg1='foo', arg2=True",
),
# job_timeout is used by enqueue for the job
("another task", "func_to_run", {"job_timeout": 180}, ""),
# timeout is NOT used by enqueue for the job (deprecated in RQ >= 1.0)
# it's passed to the function
("task4", "my_func4", {"timeout": 60}, "timeout=60"),
],
)
def test_launch_task_kwargs(user, name, func, input_kwargs, output_args):
task = user.launch_task(name, func=func, **input_kwargs)
assert task.name == name
assert task.command == f"app.tasks.{func}({output_args})"
This diff is collapsed.
......@@ -9,6 +9,7 @@ This module defines utils tests.
:license: BSD 2-Clause, see LICENSE for more details.
"""
import pytest
from pathlib import Path
from app import utils
......@@ -45,3 +46,17 @@ class TestUniqueFilename:
p = tmpdir.join("test")
p.write("Hello")
assert utils.unique_filename(p) == Path(tmpdir.join("test-1"))
@pytest.mark.parametrize(
"input,expected",
[
([], ""),
(["foo"], "foo"),
(["foo", "bar"], "foo"),
("hello", "hello"),
("", ""),
],
)
def test_attribute_to_string(input, expected):
assert utils.attribute_to_string(input) == expected