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

winupdate task tests

This commit is contained in:
sadnub 2020-09-30 21:35:46 -04:00
parent 414e7bec99
commit f72ee3c7ef
10 changed files with 190 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
api/tacticalrmm/agents/baker_recipes.py
View File

@ -1,21 +1,29 @@
from .models import Agent
from model_bakery.recipe import Recipe, seq
from model_bakery import baker
from itertools import cycle
import datetime as dt
agent = Recipe(Agent, client="Default", site="Default", hostname="TestHostname")
server_agent = Recipe(
Agent,
monitoring_mode="server",
client="Default",
site="Default",
hostname="ServerHost",
agent = Recipe(
Agent,
client="Default",
site="Default",
hostname=seq("TestHostname"),
monitoring_type=cycle(["workstation", "server"]),
)
workstation_agent = Recipe(
Agent,
monitoring_mode="server",
client="Default",
site="Default",
hostname="WorkstationHost",
server_agent = agent.extend(
monitoring_type="server",
)
workstation_agent = agent.extend(
monitoring_type="workstation",
)
online_agent = agent.extend(
last_seen=dt.datetime.now()
)
overdue_agent = agent.extend(
last_seen=dt.datetime.now(dt.timezone.utc) - dt.timedelta(minutes=6)
)

2
api/tacticalrmm/automation/views.py
View File

@ -498,4 +498,4 @@ class UpdatePatchPolicy(APIView):
def delete(self, request, patchpolicy):
get_object_or_404(WinUpdatePolicy, pk=patchpolicy).delete()
return Response("ok")
return Response("ok")

2
api/tacticalrmm/logs/admin.py
View File

@ -3,4 +3,4 @@ from django.contrib import admin
from .models import PendingAction, AuditLog
admin.site.register(PendingAction)
admin.site.register(AuditLog)
admin.site.register(AuditLog)

8
api/tacticalrmm/logs/models.py
View File

@ -168,14 +168,18 @@ class DebugLog(models.Model):
class PendingAction(models.Model):
agent = models.ForeignKey(
Agent, related_name="pendingactions", on_delete=models.CASCADE,
Agent,
related_name="pendingactions",
on_delete=models.CASCADE,
)
entry_time = models.DateTimeField(auto_now_add=True)
action_type = models.CharField(
max_length=255, choices=ACTION_TYPE_CHOICES, null=True, blank=True
)
status = models.CharField(
max_length=255, choices=STATUS_CHOICES, default="pending",
max_length=255,
choices=STATUS_CHOICES,
default="pending",
)
celery_id = models.CharField(null=True, blank=True, max_length=255)
details = models.JSONField(null=True, blank=True)

6
api/tacticalrmm/logs/tests.py
View File

@ -153,7 +153,11 @@ class TestAuditViews(TacticalTestCase):
def test_agent_pending_actions(self):
agent = baker.make_recipe("agents.agent")
pending_actions = baker.make("logs.PendingAction", agent=agent, _quantity=6,)
pending_actions = baker.make(
"logs.PendingAction",
agent=agent,
_quantity=6,
)
url = f"/logs/{agent.pk}/pendingactions/"
resp = self.client.get(url, format="json")

54
api/tacticalrmm/tacticalrmm/test.py
View File

@ -29,60 +29,6 @@ class TacticalTestCase(TestCase):
def client_setup(self):
self.client = APIClient()
def agent_setup(self):
self.agent = Agent.objects.create(
operating_system="Windows 10",
plat="windows",
plat_release="windows-Server2019",
hostname="DESKTOP-TEST123",
salt_id="aksdjaskdjs",
local_ip="10.0.25.188",
agent_id="71AHC-AA813-HH1BC-AAHH5-00013|DESKTOP-TEST123",
services=[
{
"pid": 880,
"name": "AeLookupSvc",
"status": "stopped",
"binpath": "C:\\Windows\\system32\\svchost.exe -k netsvcs",
"username": "localSystem",
"start_type": "manual",
"description": "Processes application compatibility cache requests for applications as they are launched",
"display_name": "Application Experience",
},
{
"pid": 812,
"name": "ALG",
"status": "stopped",
"binpath": "C:\\Windows\\System32\\alg.exe",
"username": "NT AUTHORITY\\LocalService",
"start_type": "manual",
"description": "Provides support for 3rd party protocol plug-ins for Internet Connection Sharing",
"display_name": "Application Layer Gateway Service",
},
],
public_ip="74.13.24.14",
total_ram=16,
used_ram=33,
disks={
"C:": {
"free": "42.3G",
"used": "17.1G",
"total": "59.5G",
"device": "C:",
"fstype": "NTFS",
"percent": 28,
}
},
boot_time=8173231.4,
logged_in_username="John",
client="Google",
site="Main Office",
monitoring_type="server",
description="Test PC",
mesh_node_id="abcdefghijklmnopAABBCCDD77443355##!!AI%@#$%#*",
last_seen=djangotime.now(),
)
# fixes tests waiting 2 minutes for mesh token to appear
@override_settings(MESH_TOKEN_KEY="123456")
def setup_coresettings(self):

42
api/tacticalrmm/winupdate/baker_recipes.py Normal file
View File

@ -0,0 +1,42 @@
from itertools import cycle
from datetime import datetime as dt
import pytz
from model_bakery.recipe import Recipe, seq
from .models import WinUpdate, WinUpdatePolicy
timezone = pytz.timezone("America/Los_Angeles")
severity = ["Critical", "Important", "Moderate", "Low", ""]
winupdate = Recipe(
WinUpdate,
kb=seq("kb0000000"),
guid=seq("12312331-123232-123-123-1123123"),
severity=cycle(severity),
)
approved_winupdate = winupdate.extend(action="approve")
winupdate_policy = Recipe(
WinUpdatePolicy,
run_time_hour=dt.now(timezone).hour,
run_time_frequency="daily",
run_time_days=[dt.now(timezone).weekday()],
)
winupdate_approve = winupdate_policy.extend(
critical="approve",
important="approve",
moderate="approve",
low="approve",
other="approve",
)
winupdate_approve_monthly = winupdate_policy.extend(
run_time_frequency="monthly",
run_time_day=dt.now(timezone).day,
critical="approve",
important="approve",
moderate="approve",
low="approve",
other="approve",
)

2
api/tacticalrmm/winupdate/models.py
View File

@ -130,4 +130,4 @@ class WinUpdatePolicy(models.Model):
# serializes the policy and returns json
from .serializers import WinUpdatePolicySerializer
return WinUpdatePolicySerializer(policy).data
return WinUpdatePolicySerializer(policy).data

9
api/tacticalrmm/winupdate/tasks.py
View File

@ -56,9 +56,9 @@ def check_agent_update_schedule_task():
# get current time in agent local time
timezone = pytz.timezone(agent.timezone)
agent_localtime_now = dt.datetime.now(timezone)
weekday = int(agent_localtime_now.strftime("%w"))
hour = int(agent_localtime_now.strftime("%-H"))
day = int(agent_localtime_now.strftime("%-d"))
weekday = agent_localtime_now.weekday()
hour = agent_localtime_now.hour
day = agent_localtime_now.day
if agent.patches_last_installed:
# get agent last installed time in local time zone
@ -82,8 +82,7 @@ def check_agent_update_schedule_task():
if patch_policy.run_time_day > 28:
months_with_30_days = [3, 6, 9, 11]
current_month = int(agent_localtime_now.strftime("%-m"))
current_month = agent_localtime_now.month
if current_month == 2:
patch_policy.run_time_day = 28
elif current_month in months_with_30_days:

112
api/tacticalrmm/winupdate/tests.py
View File

@ -1,9 +1,9 @@
from tacticalrmm.test import TacticalTestCase
from .serializers import UpdateSerializer, WinUpdateSerializer, ApprovedUpdateSerializer
from .serializers import UpdateSerializer
from model_bakery import baker
from model_bakery.recipe import foreign_key
from itertools import cycle
from unittest.mock import patch
from pprint import pprint
from .models import WinUpdate
class TestWinUpdateViews(TacticalTestCase):
@ -83,6 +83,8 @@ class TestWinUpdateViews(TacticalTestCase):
resp = self.client.get(url, format="json")
self.assertEqual(resp.status_code, 200)
self.check_not_authenticated("get", url)
def test_edit_policy(self):
url = "/winupdate/editpolicy/"
winupdate = baker.make("winupdate.WinUpdate")
@ -97,4 +99,106 @@ class TestWinUpdateViews(TacticalTestCase):
resp = self.client.patch(url, data, format="json")
self.assertEqual(resp.status_code, 200)
# TODO: add agent api to test
self.check_not_authenticated("patch", url)
class WinupdateTasks(TacticalTestCase):
def setUp(self):
self.setup_coresettings()
baker.make("clients.Site", site="Default", client__client="Default")
self.online_agents = baker.make_recipe("agents.online_agent", _quantity=2)
self.offline_agent = baker.make_recipe("agents.agent")
@patch("winupdate.tasks.check_for_updates_task.apply_async")
def test_auto_approve_task(self, check_updates_task):
from .tasks import auto_approve_updates_task
# Setup data
baker.make_recipe(
"winupdate.winupdate",
agent=cycle(
[self.online_agents[0], self.online_agents[1], self.offline_agent]
),
_quantity=20,
)
baker.make_recipe(
"winupdate.winupdate_approve",
agent=cycle(
[self.online_agents[0], self.online_agents[1], self.offline_agent]
),
_quantity=3,
)
# run task synchronously
auto_approve_updates_task()
# make sure the check_for_updates_task was run once for each online agent
self.assertEqual(check_updates_task.call_count, 2)
# check if all of the created updates were approved
winupdates = WinUpdate.objects.all()
for update in winupdates:
self.assertEqual(update.action, "approve")
@patch("agents.models.Agent.salt_api_async")
def test_check_agent_update_daily_schedule(self, agent_salt_cmd):
from .tasks import check_agent_update_schedule_task
# Setup data
# create an online agent with auto approval turned off
agent = baker.make_recipe("agents.online_agent")
baker.make("winupdate.WinUpdatePolicy", agent=agent)
# create approved winupdates
baker.make_recipe(
"winupdate.approved_winupdate",
agent=cycle(
[self.online_agents[0], self.online_agents[1], self.offline_agent]
),
_quantity=20,
)
# create daily patch policy schedules for the agents
winupdate_policy = baker.make_recipe(
"winupdate.winupdate_approve",
agent=cycle(
[self.online_agents[0], self.online_agents[1], self.offline_agent]
),
_quantity=3,
)
check_agent_update_schedule_task()
agent_salt_cmd.assert_called_with(func="win_agent.install_updates")
self.assertEquals(agent_salt_cmd.call_count, 2)
@patch("agents.models.Agent.salt_api_async")
def test_check_agent_update_monthly_schedule(self, agent_salt_cmd):
from .tasks import check_agent_update_schedule_task
# Setup data
# create an online agent with auto approval turned off
agent = baker.make_recipe("agents.online_agent")
baker.make("winupdate.WinUpdatePolicy", agent=agent)
# create approved winupdates
baker.make_recipe(
"winupdate.approved_winupdate",
agent=cycle(
[self.online_agents[0], self.online_agents[1], self.offline_agent]
),
_quantity=20,
)
# create monthly patch policy schedules for the agents
winupdate_policy = baker.make_recipe(
"winupdate.winupdate_approve_monthly",
agent=cycle(
[self.online_agents[0], self.online_agents[1], self.offline_agent]
),
_quantity=3,
)
check_agent_update_schedule_task()
agent_salt_cmd.assert_called_with(func="win_agent.install_updates")
self.assertEquals(agent_salt_cmd.call_count, 2)