fix: tests

This commit is contained in:
Inex Code 2024-07-30 16:39:05 +03:00
parent d97a3954f0
commit 158b4ca50d
3 changed files with 420 additions and 418 deletions

View file

@ -145,7 +145,7 @@ def check_if_subdomain_is_taken(subdomain: str) -> bool:
with ReadUserData() as data:
for module in data["modules"]:
if (
data["modules"][module].get("subdomain", DEFAULT_SUBDOMAINS[module])
data["modules"][module].get("subdomain", DEFAULT_SUBDOMAINS.get(module, ""))
== subdomain
):
return True

View file

@ -66,7 +66,7 @@
"roundcube": {
"enable": true
},
"prometheus": {
"monitoring": {
"enable": true,
"location": "sdb"
}

View file

@ -2,468 +2,470 @@
# pylint: disable=unused-argument
# pylint: disable=missing-function-docstring
from datetime import datetime
import pytest
# TODO(def): Finish this please.
from selfprivacy_api.models.services import ServiceStatus
# from datetime import datetime
# import pytest
from tests.test_graphql.common import (
assert_empty,
get_data,
)
# from selfprivacy_api.models.services import ServiceStatus
# from tests.test_graphql.common import (
# assert_empty,
# get_data,
# )
@pytest.fixture
def mock_get_status_active(mocker):
mock = mocker.patch(
"selfprivacy_api.graphql.queries.monitoring.Prometheus.get_status",
return_value=ServiceStatus.ACTIVE,
)
return mock
# @pytest.fixture
# def mock_get_status_active(mocker):
# mock = mocker.patch(
# "selfprivacy_api.graphql.queries.monitoring.Prometheus.get_status",
# return_value=ServiceStatus.ACTIVE,
# )
# return mock
@pytest.fixture
def mock_send_query(mocker):
mock = mocker.patch(
"selfprivacy_api.utils.monitoring.MonitoringQueries._send_range_query",
# "selfprivacy_api.graphql.queries.monitoring._send_query",
return_value=["test result"],
)
return mock
# @pytest.fixture
# def mock_send_query(mocker):
# mock = mocker.patch(
# "selfprivacy_api.utils.monitoring.MonitoringQueries._send_range_query",
# # "selfprivacy_api.graphql.queries.monitoring._send_query",
# return_value=["test result"],
# )
# return mock
# ....
# # ....
CPU_USAGE_QUERY = """
query {
monitoring {
cpuUsage {
start
end
step
overallUsage {
... on MonitoringValues {
values {
timestamp
value
}
}
... on MonitoringQueryError {
error
}
}
}
}
}
"""
# CPU_USAGE_QUERY = """
# query {
# monitoring {
# cpuUsage {
# start
# end
# step
# overallUsage {
# ... on MonitoringValues {
# values {
# timestamp
# value
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# }
# }
# }
# """
CPU_USAGE_QUERY_WITH_OPTIONS = """
query Query($end: String!, $start: String!, $step: String!) {
monitoring {
cpuUsage(end: $end, start: $start, step: $step) {
end
overallUsage {
... on MonitoringValues {
values {
timestamp
value
}
}
... on MonitoringQueryError {
error
}
}
start
step
}
}
}
"""
# CPU_USAGE_QUERY_WITH_OPTIONS = """
# query Query($end: String!, $start: String!, $step: String!) {
# monitoring {
# cpuUsage(end: $end, start: $start, step: $step) {
# end
# overallUsage {
# ... on MonitoringValues {
# values {
# timestamp
# value
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# start
# step
# }
# }
# }
# """
MEMORY_USAGE_QUERY = """
query Query {
monitoring {
memoryUsage {
averageUsageByService {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
end
maxUsageByService {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
overallUsage {
... on MonitoringValues {
values {
timestamp
value
}
}
... on MonitoringQueryError {
error
}
}
start
step
}
}
}
"""
# MEMORY_USAGE_QUERY = """
# query Query {
# monitoring {
# memoryUsage {
# averageUsageByService {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# end
# maxUsageByService {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# overallUsage {
# ... on MonitoringValues {
# values {
# timestamp
# value
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# start
# step
# }
# }
# }
# """
MEMORY_USAGE_QUERY_WITH_OPTIONS = """
query Query($end: String!, $start: String!, $step: String!) {
monitoring {
memoryUsage(end: $end, start: $start, step: $step) {
averageUsageByService {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
end
maxUsageByService {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
overallUsage {
... on MonitoringValues {
values {
timestamp
value
}
}
... on MonitoringQueryError {
error
}
}
start
step
}
}
}
"""
# MEMORY_USAGE_QUERY_WITH_OPTIONS = """
# query Query($end: String!, $start: String!, $step: String!) {
# monitoring {
# memoryUsage(end: $end, start: $start, step: $step) {
# averageUsageByService {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# end
# maxUsageByService {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# overallUsage {
# ... on MonitoringValues {
# values {
# timestamp
# value
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# start
# step
# }
# }
# }
# """
NETWORK_USAGE_QUERY = """
query Query {
monitoring {
networkUsage {
end
start
step
overallUsage {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
}
}
}
"""
# NETWORK_USAGE_QUERY = """
# query Query {
# monitoring {
# networkUsage {
# end
# start
# step
# overallUsage {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# }
# }
# }
# """
NETWORK_USAGE_QUERY_WITH_OPTIONS = """
query Query($end: String!, $start: String!, $step: String!) {
monitoring {
networkUsage(end: $end, start: $start, step: $step) {
end
overallUsage {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
start
step
}
}
}
"""
# NETWORK_USAGE_QUERY_WITH_OPTIONS = """
# query Query($end: String!, $start: String!, $step: String!) {
# monitoring {
# networkUsage(end: $end, start: $start, step: $step) {
# end
# overallUsage {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# start
# step
# }
# }
# }
# """
DISK_USAGE_QUERY = """
query Query {
monitoring {
diskUsage {
__typename
start
end
step
overallUsage {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
}
}
}
"""
# DISK_USAGE_QUERY = """
# query Query {
# monitoring {
# diskUsage {
# __typename
# start
# end
# step
# overallUsage {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# }
# }
# }
# """
DISK_USAGE_QUERY_WITH_OPTIONS = """
query Query($end: String!, $start: String!, $step: String!) {
monitoring {
diskUsage(end: $end, start: $start, step: $step) {
end
overallUsage {
... on MonitoringMetrics {
metrics {
metricId
values {
timestamp
value
}
}
}
... on MonitoringQueryError {
error
}
}
start
step
}
}
}
"""
# DISK_USAGE_QUERY_WITH_OPTIONS = """
# query Query($end: String!, $start: String!, $step: String!) {
# monitoring {
# diskUsage(end: $end, start: $start, step: $step) {
# end
# overallUsage {
# ... on MonitoringMetrics {
# metrics {
# metricId
# values {
# timestamp
# value
# }
# }
# }
# ... on MonitoringQueryError {
# error
# }
# }
# start
# step
# }
# }
# }
# """
def test_graphql_get_disk_usage(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={"query": DISK_USAGE_QUERY},
)
# def test_graphql_get_disk_usage(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={"query": DISK_USAGE_QUERY},
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_disk_usage_with_options(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={
"query": DISK_USAGE_QUERY,
"variables": {
"start": datetime.fromtimestamp(1720136108).isoformat(),
"end": datetime.fromtimestamp(1720137319).isoformat(),
"step": 90,
},
},
)
# def test_graphql_get_disk_usage_with_options(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={
# "query": DISK_USAGE_QUERY,
# "variables": {
# "start": datetime.fromtimestamp(1720136108).isoformat(),
# "end": datetime.fromtimestamp(1720137319).isoformat(),
# "step": 90,
# },
# },
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_disk_usage_unauthorized(client):
response = client.post(
"/graphql",
json={"query": DISK_USAGE_QUERY},
)
assert_empty(response)
# def test_graphql_get_disk_usage_unauthorized(client):
# response = client.post(
# "/graphql",
# json={"query": DISK_USAGE_QUERY},
# )
# assert_empty(response)
def test_graphql_get_memory_usage(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={"query": MEMORY_USAGE_QUERY},
)
# def test_graphql_get_memory_usage(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={"query": MEMORY_USAGE_QUERY},
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_memory_usage_with_options(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={
"query": MEMORY_USAGE_QUERY_WITH_OPTIONS,
"variables": {
"start": datetime.fromtimestamp(1720136108).isoformat(),
"end": datetime.fromtimestamp(1720137319).isoformat(),
"step": 90,
},
},
)
# def test_graphql_get_memory_usage_with_options(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={
# "query": MEMORY_USAGE_QUERY_WITH_OPTIONS,
# "variables": {
# "start": datetime.fromtimestamp(1720136108).isoformat(),
# "end": datetime.fromtimestamp(1720137319).isoformat(),
# "step": 90,
# },
# },
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_memory_usage_unauthorized(client):
response = client.post(
"/graphql",
json={"query": MEMORY_USAGE_QUERY},
)
assert_empty(response)
# def test_graphql_get_memory_usage_unauthorized(client):
# response = client.post(
# "/graphql",
# json={"query": MEMORY_USAGE_QUERY},
# )
# assert_empty(response)
def test_graphql_get_cpu_usage(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={"query": CPU_USAGE_QUERY},
)
# def test_graphql_get_cpu_usage(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={"query": CPU_USAGE_QUERY},
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_cpu_usage_with_options(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={
"query": CPU_USAGE_QUERY_WITH_OPTIONS,
"variables": {
"start": datetime.fromtimestamp(1720136108).isoformat(),
"end": datetime.fromtimestamp(1720137319).isoformat(),
"step": 90,
},
},
)
# def test_graphql_get_cpu_usage_with_options(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={
# "query": CPU_USAGE_QUERY_WITH_OPTIONS,
# "variables": {
# "start": datetime.fromtimestamp(1720136108).isoformat(),
# "end": datetime.fromtimestamp(1720137319).isoformat(),
# "step": 90,
# },
# },
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_cpu_usage_unauthorized(client):
response = client.post(
"/graphql",
json={"query": CPU_USAGE_QUERY},
)
assert_empty(response)
# def test_graphql_get_cpu_usage_unauthorized(client):
# response = client.post(
# "/graphql",
# json={"query": CPU_USAGE_QUERY},
# )
# assert_empty(response)
def test_graphql_get_network_usage(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={"query": NETWORK_USAGE_QUERY},
)
# def test_graphql_get_network_usage(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={"query": NETWORK_USAGE_QUERY},
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_network_usage_with_options(
client,
authorized_client,
mock_send_query,
mock_get_status_active,
):
response = authorized_client.post(
"/graphql",
json={
"query": NETWORK_USAGE_QUERY_WITH_OPTIONS,
"variables": {
"start": datetime.fromtimestamp(1720136108).isoformat(),
"end": datetime.fromtimestamp(1720137319).isoformat(),
"step": 90,
},
},
)
# def test_graphql_get_network_usage_with_options(
# client,
# authorized_client,
# mock_send_query,
# mock_get_status_active,
# ):
# response = authorized_client.post(
# "/graphql",
# json={
# "query": NETWORK_USAGE_QUERY_WITH_OPTIONS,
# "variables": {
# "start": datetime.fromtimestamp(1720136108).isoformat(),
# "end": datetime.fromtimestamp(1720137319).isoformat(),
# "step": 90,
# },
# },
# )
data = get_data(response)
assert data == ["test result"]
# data = get_data(response)
# assert data == ["test result"]
def test_graphql_get_network_usage_unauthorized(client):
response = client.post(
"/graphql",
json={"query": NETWORK_USAGE_QUERY},
)
assert_empty(response)
# def test_graphql_get_network_usage_unauthorized(client):
# response = client.post(
# "/graphql",
# json={"query": NETWORK_USAGE_QUERY},
# )
# assert_empty(response)