more tests

This commit is contained in:
wh1te909 2022-06-22 22:10:03 +00:00
parent 0986efef29
commit 3b7b5f4ec3
9 changed files with 225 additions and 172 deletions

View File

@ -48,6 +48,14 @@ jobs:
pip install setuptools==${SETUPTOOLS_VER} wheel==${WHEEL_VER}
pip install -r requirements.txt -r requirements-test.txt
- name: Codestyle black
working-directory: api/tacticalrmm
run: |
black --exclude migrations/ --check tacticalrmm
if [ $? -ne 0 ]; then
exit 1
fi
- name: Run django tests
env:
GHACTIONS: "yes"
@ -58,16 +66,7 @@ jobs:
exit 1
fi
- name: Codestyle black
working-directory: api/tacticalrmm
run: |
black --exclude migrations/ --check tacticalrmm
if [ $? -ne 0 ]; then
exit 1
fi
- uses: codecov/codecov-action@v2
- uses: codecov/codecov-action@v3
with:
directory: ./api/tacticalrmm
files: ./api/tacticalrmm/coverage.xml
verbose: true

View File

@ -1,26 +0,0 @@
[run]
source = .
[report]
show_missing = True
include = *.py
omit =
*/__pycache__/*
*/env/*
*/migrations/*
*/static/*
manage.py
*/local_settings.py
*/apps.py
*/admin.py
*/celery.py
*/wsgi.py
*/settings.py
*/baker_recipes.py
*/urls.py
*/tests.py
*/test.py
*/tests/*
checks/utils.py
*/asgi.py
*/demo_views.py

View File

@ -0,0 +1,103 @@
from unittest.mock import patch
from rest_framework.response import Response
from tacticalrmm.test import TacticalTestCase
class TestAgentUpdate(TacticalTestCase):
def setUp(self) -> None:
self.authenticate()
self.setup_coresettings()
self.setup_base_instance()
@patch("agents.utils.generate_linux_install")
@patch("knox.models.AuthToken.objects.create")
@patch("tacticalrmm.utils.generate_winagent_exe")
@patch("core.utils.token_is_valid")
@patch("agents.utils.get_agent_url")
def test_install_agent(
self,
mock_agent_url,
mock_token_valid,
mock_gen_win_exe,
mock_auth,
mock_linux_install,
):
mock_agent_url.return_value = "https://example.com"
mock_token_valid.return_value = "", False
mock_gen_win_exe.return_value = Response("ok")
mock_auth.return_value = "", "token"
mock_linux_install.return_value = Response("ok")
url = "/agents/installer/"
# test windows dynamic exe
data = {
"installMethod": "exe",
"client": self.site2.client.pk,
"site": self.site2.pk,
"expires": 24,
"agenttype": "server",
"power": 0,
"rdp": 1,
"ping": 0,
"goarch": "amd64",
"api": "https://api.example.com",
"fileName": "rmm-client-site-server.exe",
"plat": "windows",
}
r = self.client.post(url, data, format="json")
self.assertEqual(r.status_code, 200)
mock_gen_win_exe.assert_called_with(
client=self.site2.client.pk,
site=self.site2.pk,
agent_type="server",
rdp=1,
ping=0,
power=0,
goarch="amd64",
token="token",
api="https://api.example.com",
file_name="rmm-client-site-server.exe",
)
# test linux no code sign
data["plat"] = "linux"
data["installMethod"] = "bash"
data["rdp"] = 0
data["agenttype"] = "workstation"
r = self.client.post(url, data, format="json")
self.assertEqual(r.status_code, 400)
# test linux
mock_token_valid.return_value = "token123", True
r = self.client.post(url, data, format="json")
self.assertEqual(r.status_code, 200)
mock_linux_install.assert_called_with(
client=str(self.site2.client.pk),
site=str(self.site2.pk),
agent_type="workstation",
arch="amd64",
token="token",
api="https://api.example.com",
download_url="https://example.com",
)
# test manual
data["rdp"] = 1
data["installMethod"] = "manual"
r = self.client.post(url, data, format="json")
self.assertIn("rdp", r.json()["cmd"])
self.assertNotIn("power", r.json()["cmd"])
data.update({"ping": 1, "power": 1})
r = self.client.post(url, data, format="json")
self.assertIn("power", r.json()["cmd"])
self.assertIn("ping", r.json()["cmd"])
# test powershell
data["installMethod"] = "powershell"
self.assertEqual(r.status_code, 200)
self.check_not_authenticated("post", url)

View File

@ -3,6 +3,7 @@ from unittest.mock import patch
from django.conf import settings
from django.core.management import call_command
from model_bakery import baker
from packaging import version as pyver
from agents.models import Agent
from agents.tasks import auto_self_agent_update_task, send_agent_update_task
@ -203,3 +204,110 @@ class TestAgentUpdate(TacticalTestCase):
)
send_agent_update_task(agent_ids=ids, token="", force=False)
self.assertEqual(mock_update.call_count, 6)
@patch("agents.views.token_is_valid")
@patch("agents.tasks.send_agent_update_task.delay")
def test_update_agents(self, mock_update, mock_token):
mock_token.return_value = ("", False)
url = "/agents/update/"
baker.make_recipe(
"agents.online_agent",
site=self.site2,
monitoring_type=AgentMonType.SERVER,
plat=AgentPlat.WINDOWS,
version="2.3.0",
goarch=GoArch.AMD64,
_quantity=7,
)
baker.make_recipe(
"agents.online_agent",
site=self.site2,
monitoring_type=AgentMonType.SERVER,
plat=AgentPlat.WINDOWS,
version=settings.LATEST_AGENT_VER,
goarch=GoArch.AMD64,
_quantity=3,
)
baker.make_recipe(
"agents.online_agent",
site=self.site2,
monitoring_type=AgentMonType.WORKSTATION,
plat=AgentPlat.LINUX,
version="2.0.1",
goarch=GoArch.ARM32,
_quantity=9,
)
agent_ids: list[str] = list(
Agent.objects.only("agent_id").values_list("agent_id", flat=True)
)
data = {"agent_ids": agent_ids}
expected: list[str] = [
i.agent_id
for i in Agent.objects.only("agent_id", "version")
if pyver.parse(i.version) < pyver.parse(settings.LATEST_AGENT_VER)
]
r = self.client.post(url, data, format="json")
self.assertEqual(r.status_code, 200)
mock_update.assert_called_with(agent_ids=expected, token="", force=False)
self.check_not_authenticated("post", url)
@patch("agents.views.token_is_valid")
@patch("agents.tasks.send_agent_update_task.delay")
def test_agent_update_permissions(self, update_task, mock_token):
mock_token.return_value = ("", False)
agents = baker.make_recipe("agents.agent", _quantity=5)
other_agents = baker.make_recipe("agents.agent", _quantity=7)
url = f"/agents/update/"
data = {
"agent_ids": [agent.agent_id for agent in agents]
+ [agent.agent_id for agent in other_agents]
}
# test superuser access
self.check_authorized_superuser("post", url, data)
update_task.assert_called_with(
agent_ids=data["agent_ids"], token="", force=False
)
update_task.reset_mock()
user = self.create_user_with_roles([])
self.client.force_authenticate(user=user)
self.check_not_authorized("post", url, data)
update_task.assert_not_called()
user.role.can_update_agents = True
user.role.save()
self.check_authorized("post", url, data)
update_task.assert_called_with(
agent_ids=data["agent_ids"], token="", force=False
)
update_task.reset_mock()
# limit to client
# user.role.can_view_clients.set([agents[0].client])
# self.check_authorized("post", url, data)
# update_task.assert_called_with(agent_ids=[agent.agent_id for agent in agents])
# update_task.reset_mock()
# add site
# user.role.can_view_sites.set([other_agents[0].site])
# self.check_authorized("post", url, data)
# update_task.assert_called_with(agent_ids=data["agent_ids"])
# update_task.reset_mock()
# remove client permissions
# user.role.can_view_clients.clear()
# self.check_authorized("post", url, data)
# update_task.assert_called_with(
# agent_ids=[agent.agent_id for agent in other_agents]
# )

View File

@ -8,7 +8,6 @@ import pytz
from django.conf import settings
from django.utils import timezone as djangotime
from model_bakery import baker
from packaging import version as pyver
from agents.models import Agent, AgentCustomField, AgentHistory, Note
from agents.serializers import (
@ -17,8 +16,6 @@ from agents.serializers import (
AgentNoteSerializer,
AgentSerializer,
)
from agents.tasks import auto_self_agent_update_task
from logs.models import PendingAction
from tacticalrmm.constants import (
AGENT_STATUS_OFFLINE,
AGENT_STATUS_ONLINE,
@ -26,8 +23,6 @@ from tacticalrmm.constants import (
CustomFieldModel,
CustomFieldType,
EvtLogNames,
PAAction,
PAStatus,
)
from tacticalrmm.test import TacticalTestCase
from winupdate.models import WinUpdatePolicy
@ -271,40 +266,6 @@ class TestAgentViews(TacticalTestCase):
self.check_not_authenticated("get", url)
@patch("agents.tasks.send_agent_update_task.delay")
def test_update_agents(self, mock_task):
url = f"{base_url}/update/"
baker.make_recipe(
"agents.agent",
operating_system="Windows 10 Pro, 64 bit (build 19041.450)",
version=settings.LATEST_AGENT_VER,
_quantity=15,
)
baker.make_recipe(
"agents.agent",
operating_system="Windows 10 Pro, 64 bit (build 19041.450)",
version="1.3.0",
_quantity=15,
)
agent_ids: list[str] = list(
Agent.objects.only("agent_id", "version").values_list("agent_id", flat=True)
)
data = {"agent_ids": agent_ids}
expected: list[str] = [
i.agent_id
for i in Agent.objects.only("agent_id", "version")
if pyver.parse(i.version) < pyver.parse(settings.LATEST_AGENT_VER)
]
r = self.client.post(url, data, format="json")
self.assertEqual(r.status_code, 200)
mock_task.assert_called_with(agent_ids=expected)
self.check_not_authenticated("post", url)
@patch("time.sleep", return_value=None)
@patch("agents.models.Agent.nats_cmd")
def test_agent_ping(self, nats_cmd, mock_sleep):
@ -506,42 +467,6 @@ class TestAgentViews(TacticalTestCase):
self.check_not_authenticated("patch", url)
def test_install_agent(self):
url = f"{base_url}/installer/"
site = baker.make("clients.Site")
data = {
"client": site.client.pk,
"site": site.pk,
"arch": "64",
"expires": 23,
"installMethod": "manual",
"api": "https://api.example.com",
"agenttype": "server",
"rdp": 1,
"ping": 0,
"power": 0,
"fileName": "rmm-client-site-server.exe",
}
r = self.client.post(url, data, format="json")
self.assertEqual(r.status_code, 200)
data["arch"] = "64"
r = self.client.post(url, data, format="json")
self.assertIn("rdp", r.json()["cmd"])
self.assertNotIn("power", r.json()["cmd"])
data.update({"ping": 1, "power": 1})
r = self.client.post(url, data, format="json")
self.assertIn("power", r.json()["cmd"])
self.assertIn("ping", r.json()["cmd"])
data["installMethod"] = "powershell"
self.assertEqual(r.status_code, 200)
self.check_not_authenticated("post", url)
@patch("meshctrl.utils.get_login_token")
def test_meshcentral_tabs(self, mock_token):
url = f"{base_url}/{self.agent.agent_id}/meshcentral/"
@ -1131,55 +1056,6 @@ class TestAgentPermissions(TacticalTestCase):
self.check_authorized("post", url, site_data)
self.check_authorized("post", url, client_data)
@patch("agents.tasks.send_agent_update_task.delay")
def test_agent_update_permissions(self, update_task):
agents = baker.make_recipe("agents.agent", _quantity=5)
other_agents = baker.make_recipe("agents.agent", _quantity=7)
url = f"{base_url}/update/"
data = {
"agent_ids": [agent.agent_id for agent in agents]
+ [agent.agent_id for agent in other_agents]
}
# test superuser access
self.check_authorized_superuser("post", url, data)
update_task.assert_called_with(agent_ids=data["agent_ids"])
update_task.reset_mock()
user = self.create_user_with_roles([])
self.client.force_authenticate(user=user)
self.check_not_authorized("post", url, data)
update_task.assert_not_called()
user.role.can_update_agents = True
user.role.save()
self.check_authorized("post", url, data)
update_task.assert_called_with(agent_ids=data["agent_ids"])
update_task.reset_mock()
# limit to client
# user.role.can_view_clients.set([agents[0].client])
# self.check_authorized("post", url, data)
# update_task.assert_called_with(agent_ids=[agent.agent_id for agent in agents])
# update_task.reset_mock()
# add site
# user.role.can_view_sites.set([other_agents[0].site])
# self.check_authorized("post", url, data)
# update_task.assert_called_with(agent_ids=data["agent_ids"])
# update_task.reset_mock()
# remove client permissions
# user.role.can_view_clients.clear()
# self.check_authorized("post", url, data)
# update_task.assert_called_with(
# agent_ids=[agent.agent_id for agent in other_agents]
# )
def test_get_agent_version_permissions(self):
agents = baker.make_recipe("agents.agent", _quantity=5)
other_agents = baker.make_recipe("agents.agent", _quantity=7)

View File

@ -520,13 +520,6 @@ def install_agent(request):
codesign_token, is_valid = token_is_valid()
inno = f"tacticalagent-v{version}-{plat}-{goarch}.exe"
# TODO refactor this, install method should not be same as plat
if request.data["installMethod"] == AgentPlat.LINUX:
plat = AgentPlat.LINUX
else:
plat = AgentPlat.WINDOWS
download_url = get_agent_url(goarch=goarch, plat=plat, token=codesign_token)
installer_user = User.objects.filter(is_installer_user=True).first()
@ -551,7 +544,7 @@ def install_agent(request):
file_name=request.data["fileName"],
)
elif request.data["installMethod"] == AgentPlat.LINUX:
elif request.data["installMethod"] == "bash":
# TODO
# linux agents are in beta for now, only available for sponsors for testing
# remove this after it's out of beta

View File

@ -405,7 +405,7 @@ class TestClientViews(TacticalTestCase):
"ping": 0,
"rdp": 1,
"agenttype": "server",
"arch": "64",
"goarch": "amd64",
}
r = self.client.post(url, payload, format="json")

View File

@ -1,7 +1,7 @@
[pytest]
DJANGO_SETTINGS_MODULE = tacticalrmm.settings
python_files = tests.py test_*.py
addopts = --capture=tee-sys -vv --cov --cov-config=.coveragerc --cov-report=xml
addopts = --cov . --capture=tee-sys -vv
filterwarnings =
ignore::django.core.cache.CacheKeyWarning

View File

@ -31,7 +31,7 @@ class TestUtils(TacticalTestCase):
rdp=1,
ping=0,
power=0,
arch="64",
goarch="amd64",
token="abc123",
api="https://api.example.com",
file_name="rmm-client-site-server.exe",
@ -49,7 +49,7 @@ class TestUtils(TacticalTestCase):
rdp=1,
ping=0,
power=0,
arch="64",
goarch="amd64",
token="abc123",
api="https://api.example.com",
file_name="rmm-client-site-server.exe",