mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-08-18 22:45:12 +00:00
I've learned that there is no problem in grouping mutations like we do with queries. This is a big mistake from my side, now we have legacy not so conveniently placed endpoints. I've grouped all mutations, left the copies of old ones flattened in the root for backwards compatibility. We will migrate to mutation groups on client side, and backups now only use grouped mutations. Tests are updated.
240 lines
6.3 KiB
Python
240 lines
6.3 KiB
Python
# pylint: disable=redefined-outer-name
|
|
# pylint: disable=unused-argument
|
|
# pylint: disable=missing-function-docstring
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def domain_file(mocker, datadir):
|
|
mocker.patch("selfprivacy_api.utils.DOMAIN_FILE", datadir / "domain")
|
|
return datadir
|
|
|
|
|
|
class ProcessMock:
|
|
"""Mock subprocess.Popen"""
|
|
|
|
def __init__(self, args, **kwargs):
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
def communicate(): # pylint: disable=no-method-argument
|
|
return (b"", None)
|
|
|
|
returncode = 0
|
|
|
|
|
|
class BrokenServiceMock(ProcessMock):
|
|
"""Mock subprocess.Popen for broken service"""
|
|
|
|
def communicate(): # pylint: disable=no-method-argument
|
|
return (b"Testing error", None)
|
|
|
|
returncode = 3
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess_popen(mocker):
|
|
mock = mocker.patch("subprocess.Popen", autospec=True, return_value=ProcessMock)
|
|
return mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_os_chdir(mocker):
|
|
mock = mocker.patch("os.chdir", autospec=True)
|
|
return mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess_check_output(mocker):
|
|
mock = mocker.patch(
|
|
"subprocess.check_output", autospec=True, return_value=b"Testing Linux"
|
|
)
|
|
return mock
|
|
|
|
|
|
API_REBUILD_SYSTEM_MUTATION = """
|
|
mutation rebuildSystem {
|
|
system {
|
|
runSystemRebuild {
|
|
success
|
|
message
|
|
code
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_system_rebuild_unauthorized(client, mock_subprocess_popen):
|
|
"""Test system rebuild without authorization"""
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_REBUILD_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is None
|
|
assert mock_subprocess_popen.call_count == 0
|
|
|
|
|
|
def test_graphql_system_rebuild(authorized_client, mock_subprocess_popen):
|
|
"""Test system rebuild"""
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_REBUILD_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["system"]["runSystemRebuild"]["success"] is True
|
|
assert response.json()["data"]["system"]["runSystemRebuild"]["message"] is not None
|
|
assert response.json()["data"]["system"]["runSystemRebuild"]["code"] == 200
|
|
assert mock_subprocess_popen.call_count == 1
|
|
assert mock_subprocess_popen.call_args[0][0] == [
|
|
"systemctl",
|
|
"start",
|
|
"sp-nixos-rebuild.service",
|
|
]
|
|
|
|
|
|
API_UPGRADE_SYSTEM_MUTATION = """
|
|
mutation upgradeSystem {
|
|
system {
|
|
runSystemUpgrade {
|
|
success
|
|
message
|
|
code
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_system_upgrade_unauthorized(client, mock_subprocess_popen):
|
|
"""Test system upgrade without authorization"""
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_UPGRADE_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is None
|
|
assert mock_subprocess_popen.call_count == 0
|
|
|
|
|
|
def test_graphql_system_upgrade(authorized_client, mock_subprocess_popen):
|
|
"""Test system upgrade"""
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_UPGRADE_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["system"]["runSystemUpgrade"]["success"] is True
|
|
assert response.json()["data"]["system"]["runSystemUpgrade"]["message"] is not None
|
|
assert response.json()["data"]["system"]["runSystemUpgrade"]["code"] == 200
|
|
assert mock_subprocess_popen.call_count == 1
|
|
assert mock_subprocess_popen.call_args[0][0] == [
|
|
"systemctl",
|
|
"start",
|
|
"sp-nixos-upgrade.service",
|
|
]
|
|
|
|
|
|
API_ROLLBACK_SYSTEM_MUTATION = """
|
|
mutation rollbackSystem {
|
|
system {
|
|
runSystemRollback {
|
|
success
|
|
message
|
|
code
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_system_rollback_unauthorized(client, mock_subprocess_popen):
|
|
"""Test system rollback without authorization"""
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_ROLLBACK_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is None
|
|
assert mock_subprocess_popen.call_count == 0
|
|
|
|
|
|
def test_graphql_system_rollback(authorized_client, mock_subprocess_popen):
|
|
"""Test system rollback"""
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_ROLLBACK_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
assert response.json()["data"]["system"]["runSystemRollback"]["success"] is True
|
|
assert response.json()["data"]["system"]["runSystemRollback"]["message"] is not None
|
|
assert response.json()["data"]["system"]["runSystemRollback"]["code"] == 200
|
|
assert mock_subprocess_popen.call_count == 1
|
|
assert mock_subprocess_popen.call_args[0][0] == [
|
|
"systemctl",
|
|
"start",
|
|
"sp-nixos-rollback.service",
|
|
]
|
|
|
|
|
|
API_REBOOT_SYSTEM_MUTATION = """
|
|
mutation system {
|
|
system {
|
|
rebootSystem {
|
|
success
|
|
message
|
|
code
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def test_graphql_reboot_system_unauthorized(client, mock_subprocess_popen):
|
|
response = client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_REBOOT_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is None
|
|
|
|
assert mock_subprocess_popen.call_count == 0
|
|
|
|
|
|
def test_graphql_reboot_system(authorized_client, mock_subprocess_popen):
|
|
response = authorized_client.post(
|
|
"/graphql",
|
|
json={
|
|
"query": API_REBOOT_SYSTEM_MUTATION,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json().get("data") is not None
|
|
|
|
assert response.json()["data"]["system"]["rebootSystem"]["success"] is True
|
|
assert response.json()["data"]["system"]["rebootSystem"]["message"] is not None
|
|
assert response.json()["data"]["system"]["rebootSystem"]["code"] == 200
|
|
|
|
assert mock_subprocess_popen.call_count == 1
|
|
assert mock_subprocess_popen.call_args[0][0] == ["reboot"]
|