Rooba
/
tacticalrmm
mirror of https://github.com/amidaware/tacticalrmm.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

more enum

This commit is contained in:
wh1te909 2022-04-28 17:07:58 +00:00
parent 96284f9508
commit 578d5c5830
14 changed files with 104 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
api/tacticalrmm/accounts/management/commands/create_installer_user.py
View File

@ -8,16 +8,16 @@ from accounts.models import User
class Command(BaseCommand):
help = "Creates the installer user"
def handle(self, *args, **kwargs):
def handle(self, *args, **kwargs): # type: ignore
self.stdout.write("Checking if installer user has been created...")
if User.objects.filter(is_installer_user=True).exists():
self.stdout.write("Installer user already exists")
return
User.objects.create_user( # type: ignore
User.objects.create_user(
username=uuid.uuid4().hex,
is_installer_user=True,
password=User.objects.make_random_password(60), # type: ignore
password=User.objects.make_random_password(60),
block_dashboard_login=True,
)
self.stdout.write("Installer user has been created")

2
api/tacticalrmm/accounts/management/commands/delete_tokens.py
View File

@ -6,7 +6,7 @@ from knox.models import AuthToken
class Command(BaseCommand):
help = "Deletes all knox web tokens"
def handle(self, *args, **kwargs):
def handle(self, *args, **kwargs): # type: ignore
# only delete web tokens, not any generated by the installer or deployments
dont_delete = djangotime.now() + djangotime.timedelta(hours=23)
tokens = AuthToken.objects.exclude(deploytokens__isnull=False).filter(

24
api/tacticalrmm/agents/management/commands/fake_agents.py
View File

@ -17,7 +17,7 @@ from clients.models import Client, Site
from logs.models import AuditLog, PendingAction
from scripts.models import Script
from software.models import InstalledSoftware
from tacticalrmm.constants import CheckType, PAAction
from tacticalrmm.constants import CheckType, PAAction, CheckStatus
from tacticalrmm.demo_data import (
check_network_loc_aware_ps1,
check_storage_pool_health_ps1,
@ -401,7 +401,7 @@ class Command(BaseCommand):
check_result1 = CheckResult()
check_result1.agent = agent
check_result1.assigned_check = check1
check_result1.status = "passing"
check_result1.status = CheckStatus.PASSING
check_result1.last_run = django_now
check_result1.more_info = "Total: 498.7GB, Free: 287.4GB"
check_result1.save()
@ -431,12 +431,12 @@ class Command(BaseCommand):
if site in sites5:
check2.name = "Synology NAS"
check2.alert_severity = "error"
check_result2.status = "failing"
check_result2.status = CheckStatus.FAILING
check2.ip = "172.17.14.26"
check_result2.more_info = ping_fail_output
else:
check2.name = "Google"
check_result2.status = "passing"
check_result2.status = CheckStatus.PASSING
check2.ip = "8.8.8.8"
check_result2.more_info = ping_success_output
@ -469,7 +469,7 @@ class Command(BaseCommand):
check_result3 = CheckResult()
check_result3.agent = agent
check_result3.assigned_check = check3
check_result3.status = "passing"
check_result3.status = CheckStatus.PASSING
check_result3.last_run = django_now
check_result3.history = [
15,
@ -510,7 +510,7 @@ class Command(BaseCommand):
check_result4 = CheckResult()
check_result4.agent = agent
check_result4.assigned_check = check4
check_result4.status = "passing"
check_result4.status = CheckStatus.PASSING
check_result4.last_run = django_now
check_result4.history = [34, 34, 35, 36, 34, 34, 34, 34, 34, 34]
check_result4.save()
@ -539,7 +539,7 @@ class Command(BaseCommand):
check_result5 = CheckResult()
check_result5.agent = agent
check_result5.assigned_check = check5
check_result5.status = "passing"
check_result5.status = CheckStatus.PASSING
check_result5.last_run = django_now
check_result5.retcode = 0
check_result5.execution_time = "4.0000"
@ -569,7 +569,7 @@ class Command(BaseCommand):
check_result6 = CheckResult()
check_result6.agent = agent
check_result6.assigned_check = check6
check_result6.status = "passing"
check_result6.status = CheckStatus.PASSING
check_result6.last_run = django_now
check_result6.retcode = 0
check_result6.execution_time = "4.0000"
@ -685,7 +685,7 @@ class Command(BaseCommand):
check_result7 = CheckResult()
check_result7.assigned_check = check7
check_result7.agent = agent
check_result7.status = "passing"
check_result7.status = CheckStatus.PASSING
check_result7.last_run = django_now
check_result7.retcode = 0
check_result7.execution_time = "3.1337"
@ -716,7 +716,7 @@ class Command(BaseCommand):
check_result8 = CheckResult()
check_result8.assigned_check = check8
check_result8.agent = agent
check_result8.status = "passing"
check_result8.status = CheckStatus.PASSING
check_result8.last_run = django_now
check_result8.more_info = "Status RUNNING"
check_result8.save()
@ -754,10 +754,10 @@ class Command(BaseCommand):
check_result9.last_run = django_now
if site in sites5:
check_result9.extra_details = eventlog_check_fail_data
check_result9.status = "failing"
check_result9.status = CheckStatus.FAILING
else:
check_result9.extra_details = {"log": []}
check_result9.status = "passing"
check_result9.status = CheckStatus.PASSING
check9.save()
check_result9.save()

6
api/tacticalrmm/agents/models.py
View File

@ -19,7 +19,7 @@ from packaging import version as pyver
from core.models import TZ_CHOICES
from core.utils import get_core_settings
from logs.models import BaseAuditModel, DebugLog
from tacticalrmm.constants import ONLINE_AGENTS, CheckType
from tacticalrmm.constants import ONLINE_AGENTS, CheckType, CheckStatus
from tacticalrmm.models import PermissionQuerySet
if TYPE_CHECKING:
@ -167,12 +167,12 @@ class Agent(BaseAuditModel):
if (
not hasattr(check.check_result, "status")
or isinstance(check.check_result, CheckResult)
and check.check_result.status == "passing"
and check.check_result.status == CheckStatus.PASSING
):
passing += 1
elif (
isinstance(check.check_result, CheckResult)
and check.check_result.status == "failing"
and check.check_result.status == CheckStatus.FAILING
):
alert_severity = (
check.check_result.alert_severity

4
api/tacticalrmm/agents/tasks.py
View File

@ -14,7 +14,7 @@ from core.utils import get_core_settings
from logs.models import DebugLog, PendingAction
from scripts.models import Script
from tacticalrmm.celery import app
from tacticalrmm.constants import PAAction, PAStatus
from tacticalrmm.constants import PAAction, PAStatus, CheckStatus
def agent_update(agent_id: str, force: bool = False) -> str:
@ -290,7 +290,7 @@ def clear_faults_task(older_than_days: int) -> None:
for check in agent.get_checks_with_policies():
# reset check status
if check.check_result:
check.check_result.status = "passing"
check.check_result.status = CheckStatus.PASSING
check.check_result.save(update_fields=["status"])
if check.alert.filter(agent=agent, resolved=False).exists():
alert = Alert.create_or_return_check_alert(check, agent=agent)

3
api/tacticalrmm/alerts/tests.py
View File

@ -11,6 +11,7 @@ from autotasks.models import TaskResult
from core.tasks import cache_db_fields_task, handle_resolved_stuff
from core.utils import get_core_settings
from tacticalrmm.test import TacticalTestCase
from tacticalrmm.constants import CheckStatus
from .models import Alert, AlertTemplate
from .serializers import (
@ -841,7 +842,7 @@ class TestAlertTasks(TacticalTestCase):
# test agent with check that has alert settings
check_agent_result.alert_severity = "warning"
check_agent_result.status = "failing"
check_agent_result.status = CheckStatus.FAILING
Alert.handle_alert_failure(check_agent_result)

14
api/tacticalrmm/apiv3/views.py
View File

@ -25,7 +25,7 @@ from core.utils import (
)
from logs.models import DebugLog, PendingAction
from software.models import InstalledSoftware
from tacticalrmm.constants import MeshAgentIdent, PAStatus
from tacticalrmm.constants import MeshAgentIdent, PAStatus, CheckStatus
from tacticalrmm.helpers import notify_error
from tacticalrmm.utils import reload_nats
from winupdate.models import WinUpdate, WinUpdatePolicy
@ -246,8 +246,8 @@ class CheckRunner(APIView):
check_result.save()
status = check_result.handle_check(request.data)
if status == "failing" and check.assignedtasks.exists(): # type: ignore
for task in check.assignedtasks.all(): # type: ignore
if status == CheckStatus.FAILING and check.assignedtasks.exists():
for task in check.assignedtasks.all():
if task.enabled:
if task.policy:
task.run_win_task(agent)
@ -309,11 +309,11 @@ class TaskRunner(APIView):
task_result.save_collector_results()
status = "passing"
status = CheckStatus.PASSING
else:
status = "failing"
status = CheckStatus.FAILING
else:
status = "failing" if task_result.retcode != 0 else "passing"
status = CheckStatus.FAILING if task_result.retcode != 0 else CheckStatus.PASSING
if task_result:
task_result.status = status
@ -322,7 +322,7 @@ class TaskRunner(APIView):
task_result.status = status
task.save(update_fields=["status"])
if status == "passing":
if status == CheckStatus.PASSING:
if Alert.create_or_return_task_alert(task, agent=agent, skip_create=True):
Alert.handle_alert_resolve(task_result)
else:

50
api/tacticalrmm/checks/models.py
View File

@ -13,6 +13,7 @@ from tacticalrmm.constants import (
CHECKS_NON_EDITABLE_FIELDS,
POLICY_CHECK_FIELDS_TO_COPY,
CheckType,
CheckStatus,
)
from tacticalrmm.models import PermissionQuerySet
@ -20,11 +21,6 @@ if TYPE_CHECKING:
from alerts.models import Alert, AlertTemplate # pragma: no cover
from automation.models import Policy # pragma: no cover
CHECK_STATUS_CHOICES = [
("passing", "Passing"),
("failing", "Failing"),
("pending", "Pending"),
]
EVT_LOG_NAME_CHOICES = [
("Application", "Application"),
@ -325,7 +321,7 @@ class CheckResult(models.Model):
on_delete=models.CASCADE,
)
status = models.CharField(
max_length=100, choices=CHECK_STATUS_CHOICES, default="pending"
max_length=100, choices=CheckStatus.choices, default=CheckStatus.PENDING
)
# for memory, diskspace, script, and cpu checks where severity changes
alert_severity = models.CharField(
@ -402,13 +398,13 @@ class CheckResult(models.Model):
avg = int(mean(self.history))
if check.error_threshold and avg > check.error_threshold:
self.status = "failing"
self.status = CheckStatus.FAILING
self.alert_severity = "error"
elif check.warning_threshold and avg > check.warning_threshold:
self.status = "failing"
self.status = CheckStatus.FAILING
self.alert_severity = "warning"
else:
self.status = "passing"
self.status = CheckStatus.PASSING
# add check history
check.add_check_history(data["percent"], self.agent.agent_id)
@ -421,24 +417,24 @@ class CheckResult(models.Model):
check.error_threshold
and (100 - percent_used) < check.error_threshold
):
self.status = "failing"
self.status = CheckStatus.FAILING
self.alert_severity = "error"
elif (
check.warning_threshold
and (100 - percent_used) < check.warning_threshold
):
self.status = "failing"
self.status = CheckStatus.FAILING
self.alert_severity = "warning"
else:
self.status = "passing"
self.status = CheckStatus.PASSING
self.more_info = data["more_info"]
# add check history
check.add_check_history(100 - percent_used, self.agent.agent_id)
else:
self.status = "failing"
self.status = CheckStatus.FAILING
self.alert_severity = "error"
self.more_info = f"Disk {check.disk} does not exist"
@ -453,15 +449,15 @@ class CheckResult(models.Model):
if data["retcode"] in check.info_return_codes:
self.alert_severity = "info"
self.status = "failing"
self.status = CheckStatus.FAILING
elif data["retcode"] in check.warning_return_codes:
self.alert_severity = "warning"
self.status = "failing"
self.status = CheckStatus.FAILING
elif data["retcode"] != 0:
self.status = "failing"
self.status = CheckStatus.FAILING
self.alert_severity = "error"
else:
self.status = "passing"
self.status = CheckStatus.PASSING
self.save(
update_fields=[
@ -474,7 +470,7 @@ class CheckResult(models.Model):
# add check history
check.add_check_history(
1 if self.status == "failing" else 0,
1 if self.status == CheckStatus.FAILING else 0,
self.agent.agent_id,
{
"retcode": data["retcode"],
@ -491,7 +487,7 @@ class CheckResult(models.Model):
self.save(update_fields=["more_info"])
check.add_check_history(
1 if self.status == "failing" else 0,
1 if self.status == CheckStatus.FAILING else 0,
self.agent.agent_id,
self.more_info[:60],
)
@ -503,7 +499,7 @@ class CheckResult(models.Model):
self.save(update_fields=["more_info"])
check.add_check_history(
1 if self.status == "failing" else 0,
1 if self.status == CheckStatus.FAILING else 0,
self.agent.agent_id,
self.more_info[:60],
)
@ -512,34 +508,34 @@ class CheckResult(models.Model):
log = data["log"]
if check.fail_when == "contains":
if log and len(log) >= check.number_of_events_b4_alert:
self.status = "failing"
self.status = CheckStatus.FAILING
else:
self.status = "passing"
self.status = CheckStatus.PASSING
elif check.fail_when == "not_contains":
if log and len(log) >= check.number_of_events_b4_alert:
self.status = "passing"
self.status = CheckStatus.PASSING
else:
self.status = "failing"
self.status = CheckStatus.FAILING
self.extra_details = {"log": log}
self.save(update_fields=["extra_details"])
check.add_check_history(
1 if self.status == "failing" else 0,
1 if self.status == CheckStatus.FAILING else 0,
self.agent.agent_id,
"Events Found:" + str(len(self.extra_details["log"])),
)
# handle status
if self.status == "failing":
if self.status == CheckStatus.FAILING:
self.fail_count += 1
self.save(update_fields=["status", "fail_count", "alert_severity"])
if self.fail_count >= check.fails_b4_alert:
Alert.handle_alert_failure(self)
elif self.status == "passing":
elif self.status == CheckStatus.PASSING:
self.fail_count = 0
self.save(update_fields=["status", "fail_count", "alert_severity"])
if Alert.objects.filter(

72
api/tacticalrmm/checks/tests.py
View File

@ -6,7 +6,7 @@ from django.utils import timezone as djangotime
from model_bakery import baker
from checks.models import CheckHistory, CheckResult
from tacticalrmm.constants import CheckType
from tacticalrmm.constants import CheckType, CheckStatus
from tacticalrmm.test import TacticalTestCase
from .serializers import CheckSerializer
@ -348,7 +348,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test passing
@ -366,7 +366,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
# test failing info
check.info_return_codes = [20, 30, 50]
@ -386,7 +386,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "info")
# test failing warning
@ -407,7 +407,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
def test_handle_diskspace_check(self):
@ -436,7 +436,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
# test error failure
@ -455,7 +455,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test disk not exist
@ -466,7 +466,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test warning threshold 0
@ -487,7 +487,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test error threshold 0
@ -508,7 +508,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
# test passing
@ -527,7 +527,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
def test_handle_cpuload_check(self):
url = "/api/v3/checkrunner/"
@ -546,7 +546,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
# test failing error
@ -560,7 +560,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test passing
@ -574,7 +574,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
# test warning threshold 0
check.warning_threshold = 0
@ -589,7 +589,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test error threshold 0
@ -606,7 +606,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
def test_handle_memory_check(self):
@ -626,7 +626,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
# test failing error
@ -640,7 +640,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test passing
@ -654,7 +654,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
# test warning threshold 0
check.warning_threshold = 0
@ -669,7 +669,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "error")
# test error threshold 0
@ -686,7 +686,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check_result.alert_severity, "warning")
def test_handle_ping_check(self):
@ -700,7 +700,7 @@ class TestCheckTasks(TacticalTestCase):
data = {
"id": check.id,
"agent_id": self.agent.agent_id,
"status": "failing",
"status": CheckStatus.FAILING,
"output": "reply from a.com",
}
@ -708,7 +708,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check.alert_severity, "info")
# test failing warning
@ -719,7 +719,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check.alert_severity, "warning")
# test failing error
@ -730,7 +730,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check.alert_severity, "error")
# test failing error
@ -738,14 +738,14 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check.alert_severity, "error")
# test passing
data = {
"id": check.id,
"agent_id": self.agent.agent_id,
"status": "passing",
"status": CheckStatus.PASSING,
"output": "reply from a.com",
}
@ -753,7 +753,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
@patch("agents.models.Agent.nats_cmd")
def test_handle_winsvc_check(self, nats_cmd):
@ -767,7 +767,7 @@ class TestCheckTasks(TacticalTestCase):
data = {
"id": check.id,
"agent_id": self.agent.agent_id,
"status": "passing",
"status": CheckStatus.PASSING,
"more_info": "ok",
}
@ -775,13 +775,13 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
# test failing
data = {
"id": check.id,
"agent_id": self.agent.agent_id,
"status": "failing",
"status": CheckStatus.FAILING,
"more_info": "ok",
}
@ -789,7 +789,7 @@ class TestCheckTasks(TacticalTestCase):
self.assertEqual(resp.status_code, 200)
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check.alert_severity, "info")
def test_handle_eventlog_check(self):
@ -844,7 +844,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check.alert_severity, "warning")
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
# test passing when contains
resp = self.client.patch(url, no_logs_data, format="json")
@ -852,7 +852,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
# test failing when not contains and message and source
check.fail_when = "not_contains"
@ -864,7 +864,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "failing")
self.assertEqual(check_result.status, CheckStatus.FAILING)
self.assertEqual(check.alert_severity, "error")
# test passing when contains with source and message
@ -873,7 +873,7 @@ class TestCheckTasks(TacticalTestCase):
check_result = CheckResult.objects.get(assigned_check=check, agent=self.agent)
self.assertEqual(check_result.status, "passing")
self.assertEqual(check_result.status, CheckStatus.PASSING)
class TestCheckPermissions(TacticalTestCase):

4
api/tacticalrmm/checks/views.py
View File

@ -13,7 +13,7 @@ from rest_framework.views import APIView
from agents.models import Agent
from alerts.models import Alert
from automation.models import Policy
from tacticalrmm.constants import CheckType
from tacticalrmm.constants import CheckType, CheckStatus
from tacticalrmm.helpers import notify_error
from tacticalrmm.permissions import _has_perm_on_agent
@ -117,7 +117,7 @@ class ResetCheck(APIView):
if result.agent and not _has_perm_on_agent(request.user, result.agent.agent_id):
raise PermissionDenied()
result.status = "passing"
result.status = CheckStatus.PASSING
result.save()
# resolve any alerts that are open

8
api/tacticalrmm/core/management/commands/check_mesh.py
View File

@ -9,16 +9,16 @@ from core.utils import get_core_settings, get_mesh_device_id, get_mesh_ws_url
class Command(BaseCommand):
help = "Mesh troubleshooting script"
def _success(self, *args):
def _success(self, *args) -> None:
self.stdout.write(self.style.SUCCESS(" ".join(args)))
def _error(self, *args):
def _error(self, *args) -> None:
self.stdout.write(self.style.ERROR(" ".join(args)))
def _warning(self, *args):
def _warning(self, *args) -> None:
self.stdout.write(self.style.WARNING(" ".join(args)))
def handle(self, *args, **kwargs):
def handle(self, *args, **kwargs) -> None:
core = get_core_settings()
self._warning("Mesh site:", core.mesh_site)

4
api/tacticalrmm/tacticalrmm/constants.py
View File

@ -18,6 +18,10 @@ class MeshAgentIdent(Enum):
CORESETTINGS_CACHE_KEY = "core_settings"
ROLE_CACHE_PREFIX = "role_"
class CheckStatus(models.TextChoices):
PASSING = "passing", "Passing"
FAILING = "failing", "Failing"
PENDING = "pending", "Pending"
class PAStatus(models.TextChoices):
PENDING = "pending", "Pending"

2
api/tacticalrmm/tacticalrmm/urls.py
View File

@ -60,5 +60,5 @@ if getattr(settings, "SWAGGER_ENABLED", False):
)
ws_urlpatterns = [
path("ws/dashinfo/", DashInfo.as_asgi()), # type: ignore
path("ws/dashinfo/", DashInfo.as_asgi()),
]

5
mypy.ini
View File

@ -13,4 +13,7 @@ plugins =
mypy_drf_plugin.main
[mypy.plugins.django-stubs]
django_settings_module = tacticalrmm.settings
django_settings_module = tacticalrmm.settings
[mypy-knox.*]
ignore_missing_imports = True