wehub-resource-sync
368 lines
13 KiB
Python
368 lines
13 KiB
Python
# Copyright (C) 2022 Intel Corporation
|
|
# Copyright (C) CVAT.ai Corporation
|
|
#
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
import csv
|
|
import json
|
|
import uuid
|
|
from collections import Counter
|
|
from datetime import datetime, timedelta, timezone
|
|
from http import HTTPStatus
|
|
from io import StringIO
|
|
from time import sleep
|
|
|
|
import pytest
|
|
from dateutil import parser as datetime_parser
|
|
|
|
import shared.utils.s3 as s3
|
|
from shared.utils.config import delete_method, get_method, make_api_client, server_get
|
|
from shared.utils.helpers import generate_image_files
|
|
|
|
from .utils import create_task, export_events
|
|
|
|
|
|
class TestGetAnalytics:
|
|
endpoint = "analytics"
|
|
|
|
def _test_can_see(self, user):
|
|
response = server_get(user, self.endpoint)
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
def _test_cannot_see(self, user):
|
|
response = server_get(user, self.endpoint)
|
|
|
|
assert response.status_code == HTTPStatus.FORBIDDEN
|
|
|
|
@pytest.mark.parametrize(
|
|
"conditions, is_allow",
|
|
[
|
|
(dict(privilege="admin"), True),
|
|
(dict(privilege="worker", has_analytics_access=False), False),
|
|
(dict(privilege="worker", has_analytics_access=True), True),
|
|
(dict(privilege="user", has_analytics_access=False), False),
|
|
(dict(privilege="user", has_analytics_access=True), True),
|
|
],
|
|
)
|
|
def test_can_see(self, conditions, is_allow, find_users):
|
|
user = find_users(**conditions)[0]["username"]
|
|
|
|
if is_allow:
|
|
self._test_can_see(user)
|
|
else:
|
|
self._test_cannot_see(user)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_class")
|
|
class TestGetAuditEvents:
|
|
_USERNAME = "admin1"
|
|
|
|
@staticmethod
|
|
def _create_project(user, spec, **kwargs):
|
|
with make_api_client(user) as api_client:
|
|
project, response = api_client.projects_api.create(spec, **kwargs)
|
|
assert response.status == HTTPStatus.CREATED
|
|
return project.id, response.headers.get("X-Request-Id")
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup(self, restore_clickhouse_db_per_function, restore_redis_inmem_per_function):
|
|
project_spec = {
|
|
"name": f"Test project created by {self._USERNAME}",
|
|
"labels": [
|
|
{
|
|
"name": "car",
|
|
"color": "#ff00ff",
|
|
"attributes": [
|
|
{
|
|
"name": "a",
|
|
"mutable": True,
|
|
"input_type": "number",
|
|
"default_value": "5",
|
|
"values": ["4", "5", "6"],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
self.project_id, project_request_id = TestGetAuditEvents._create_project(
|
|
self._USERNAME, project_spec
|
|
)
|
|
task_spec = {
|
|
"name": f"test {self._USERNAME} to create a task",
|
|
"segment_size": 2,
|
|
"project_id": self.project_id,
|
|
}
|
|
task_ids = [
|
|
create_task(
|
|
self._USERNAME,
|
|
task_spec,
|
|
{
|
|
"image_quality": 10,
|
|
"client_files": generate_image_files(3),
|
|
},
|
|
),
|
|
create_task(
|
|
self._USERNAME,
|
|
task_spec,
|
|
{
|
|
"image_quality": 10,
|
|
"client_files": generate_image_files(3),
|
|
},
|
|
),
|
|
]
|
|
|
|
self.task_ids = [t[0] for t in task_ids]
|
|
|
|
assert project_request_id is not None
|
|
assert all(t[1] is not None for t in task_ids)
|
|
|
|
# Events are pushed to ClickHouse asynchronously,
|
|
# so we must wait until the expected number of events for every scope is present.
|
|
# Items: (filters, expected_count)
|
|
event_filters = [
|
|
(
|
|
(
|
|
(lambda e: json.loads(e["payload"])["request"]["id"], [project_request_id]),
|
|
("scope", ["create:project"]),
|
|
),
|
|
1,
|
|
),
|
|
]
|
|
for task_id, task_request_id in task_ids:
|
|
event_filters.extend(
|
|
(
|
|
(
|
|
(
|
|
(
|
|
lambda e: json.loads(e["payload"])["request"]["id"],
|
|
[task_request_id],
|
|
),
|
|
("scope", ["create:task"]),
|
|
),
|
|
1,
|
|
),
|
|
(
|
|
(
|
|
("task_id", [str(task_id)]),
|
|
("scope", ["create:job"]),
|
|
),
|
|
2,
|
|
),
|
|
)
|
|
)
|
|
self._wait_for_request_ids(event_filters)
|
|
|
|
def _wait_for_request_ids(self, event_filters):
|
|
MAX_RETRIES = 5
|
|
SLEEP_INTERVAL = 2
|
|
while MAX_RETRIES > 0:
|
|
data = self._test_get_audit_logs_as_csv()
|
|
events = self._csv_to_dict(data)
|
|
if all(
|
|
len(self._filter_events(events, filters)) >= expected_count
|
|
for filters, expected_count in event_filters
|
|
):
|
|
break
|
|
MAX_RETRIES -= 1
|
|
sleep(SLEEP_INTERVAL)
|
|
else:
|
|
assert False, "Could not wait for expected request IDs"
|
|
|
|
@staticmethod
|
|
def _csv_to_dict(csv_data):
|
|
res = []
|
|
with StringIO(csv_data.decode()) as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
res.append(row)
|
|
|
|
return res
|
|
|
|
@staticmethod
|
|
def _filter_events(events, filters):
|
|
res = []
|
|
get_value = lambda getter, e: getter(e) if callable(getter) else e.get(getter, None)
|
|
for e in events:
|
|
if all(get_value(getter, e) in expected_values for getter, expected_values in filters):
|
|
res.append(e)
|
|
|
|
return res
|
|
|
|
def _test_get_audit_logs_as_csv(self, *, api_version: int = 2, **kwargs):
|
|
with make_api_client(self._USERNAME) as api_client:
|
|
return export_events(api_client, api_version=api_version, **kwargs)
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_entry_to_time_interval(self, api_version: int):
|
|
now = datetime.now(timezone.utc)
|
|
to_datetime = now
|
|
from_datetime = now - timedelta(minutes=3)
|
|
|
|
query_params = {
|
|
"_from": from_datetime.isoformat(),
|
|
"to": to_datetime.isoformat(),
|
|
}
|
|
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
assert len(events)
|
|
|
|
for event in events:
|
|
event_timestamp = datetime_parser.isoparse(event["timestamp"])
|
|
assert from_datetime <= event_timestamp <= to_datetime
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_filter_by_project(self, api_version: int):
|
|
query_params = {
|
|
"project_id": self.project_id,
|
|
}
|
|
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
|
|
filtered_events = self._filter_events(events, [("project_id", [str(self.project_id)])])
|
|
assert len(filtered_events)
|
|
assert len(events) == len(filtered_events)
|
|
|
|
event_count = Counter([e["scope"] for e in filtered_events])
|
|
assert event_count["create:project"] == 1
|
|
assert event_count["create:task"] == 2
|
|
assert event_count["create:job"] == 4
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_filter_by_task(self, api_version: int):
|
|
for task_id in self.task_ids:
|
|
query_params = {
|
|
"task_id": task_id,
|
|
}
|
|
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
|
|
filtered_events = self._filter_events(events, [("task_id", [str(task_id)])])
|
|
assert len(filtered_events)
|
|
assert len(events) == len(filtered_events)
|
|
|
|
event_count = Counter([e["scope"] for e in filtered_events])
|
|
assert event_count["create:task"] == 1
|
|
assert event_count["create:job"] == 2
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_filter_by_non_existent_project(self, api_version: int):
|
|
query_params = {
|
|
"project_id": self.project_id + 100,
|
|
}
|
|
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
assert len(events) == 0
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_user_and_request_id_not_empty(self, api_version: int):
|
|
query_params = {
|
|
"project_id": self.project_id,
|
|
}
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
|
|
for event in events:
|
|
assert event["user_id"]
|
|
assert event["user_name"]
|
|
assert event["user_email"]
|
|
|
|
payload = json.loads(event["payload"])
|
|
request_id = payload["request"]["id"]
|
|
assert request_id
|
|
uuid.UUID(request_id)
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_exported_events_do_not_contain_remote_addr(self, api_version: int):
|
|
query_params = {
|
|
"project_id": self.project_id,
|
|
}
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
|
|
assert len(events)
|
|
for event in events:
|
|
assert "remote_addr" not in event
|
|
|
|
@pytest.mark.parametrize("api_version", [1, 2])
|
|
def test_delete_project(self, api_version: int):
|
|
response = delete_method("admin1", f"projects/{self.project_id}")
|
|
assert response.status_code == HTTPStatus.NO_CONTENT
|
|
|
|
request_id = response.headers.get("X-Request-Id")
|
|
# Items: (filters, expected_count). The cascade delete emits one event per
|
|
# affected object under the same request id, so wait for all of them.
|
|
event_filters = (
|
|
(
|
|
(
|
|
(lambda e: json.loads(e["payload"])["request"]["id"], [request_id]),
|
|
("scope", ["delete:project"]),
|
|
),
|
|
1,
|
|
),
|
|
(
|
|
(
|
|
(lambda e: json.loads(e["payload"])["request"]["id"], [request_id]),
|
|
("scope", ["delete:task"]),
|
|
),
|
|
2,
|
|
),
|
|
(
|
|
(
|
|
(lambda e: json.loads(e["payload"])["request"]["id"], [request_id]),
|
|
("scope", ["delete:job"]),
|
|
),
|
|
4,
|
|
),
|
|
)
|
|
|
|
self._wait_for_request_ids(event_filters)
|
|
|
|
query_params = {
|
|
"project_id": self.project_id,
|
|
}
|
|
|
|
data = self._test_get_audit_logs_as_csv(api_version=api_version, **query_params)
|
|
events = self._csv_to_dict(data)
|
|
|
|
filtered_events = self._filter_events(events, [("project_id", [str(self.project_id)])])
|
|
assert len(filtered_events)
|
|
assert len(events) == len(filtered_events)
|
|
|
|
event_count = Counter([e["scope"] for e in filtered_events])
|
|
assert event_count["delete:project"] == 1
|
|
assert event_count["delete:task"] == 2
|
|
assert event_count["delete:job"] == 4
|
|
|
|
@pytest.mark.with_external_services
|
|
@pytest.mark.parametrize("api_version, allowed", [(1, False), (2, True)])
|
|
@pytest.mark.parametrize("cloud_storage_id", [3]) # import/export bucket
|
|
def test_export_to_cloud(
|
|
self, api_version: int, allowed: bool, cloud_storage_id: int, cloud_storages
|
|
):
|
|
query_params = {
|
|
"api_version": api_version,
|
|
"location": "cloud_storage",
|
|
"cloud_storage_id": cloud_storage_id,
|
|
"filename": "test.csv",
|
|
"task_id": self.task_ids[0],
|
|
}
|
|
if allowed:
|
|
data = self._test_get_audit_logs_as_csv(**query_params)
|
|
assert data is None
|
|
s3_client = s3.make_client(bucket=cloud_storages[cloud_storage_id]["resource"])
|
|
data = s3_client.download_fileobj(query_params["filename"])
|
|
events = self._csv_to_dict(data)
|
|
assert len(events)
|
|
else:
|
|
response = get_method(self._USERNAME, "events", **query_params)
|
|
assert response.status_code == HTTPStatus.BAD_REQUEST
|
|
assert (
|
|
response.json()[0]
|
|
== "This endpoint does not support exporting events to cloud storage"
|
|
)
|