Merge pull request #30 from jtdub/core-refactor

rename home app to core app

.pylintrc

@@ -1,5 +1,5 @@
 [MASTER]
-ignore=migrations
+ignore=migrations,tests
 
 [MESSAGES CONTROL]
 disable=no-member,too-few-public-methods,too-many-ancestors, duplicate-code

netopsio/core/__init__.py

@@ -0,0 +1 @@
+"""Netops.io core app."""

netopsio/core/admin.py

@@ -0,0 +1,14 @@
+"""Register core models in admin panel."""
+
+from django.contrib import admin
+
+from core.models import RequestLog
+
+
+class RequestLogAdmin(admin.ModelAdmin):
+    """Request Log Admin Site Fields."""
+
+    list_display = ("ip", "app", "date")
+
+
+admin.site.register(RequestLog, RequestLogAdmin)

netopsio/core/api/__init__.py

@@ -0,0 +1 @@
+"""Core application API."""

netopsio/home/serializers.py → netopsio/core/api/serializers.py

@@ -1,6 +1,8 @@
-"""Home App and Worker Serializers."""
-from rest_framework import serializers
+"""Core App and Worker Serializers."""
 from django_celery_results.models import TaskResult
+from rest_framework import serializers
+
+from core import models
 
 
 class TaskResultSerializer(serializers.HyperlinkedModelSerializer):
@@ -24,3 +26,19 @@ class Meta:
             "date_done",
             "url",
         ]
+
+
+class TaskSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """Ping Serializer."""
+
+    task_id = serializers.CharField(max_length=512)
+
+
+class RequestLogSerializer(serializers.ModelSerializer):
+    """RequestLog Serializer."""
+
+    class Meta:
+        """RequestLog Serializer Meta."""
+
+        model = models.RequestLog
+        fields = ["id", "date", "task_id", "ip", "app", "result", "url"]

netopsio/core/api/urls.py

@@ -0,0 +1,12 @@
+"""Netops.io core api urls."""
+
+from rest_framework.routers import DefaultRouter
+
+from core.api import views
+
+router = DefaultRouter()
+router.register(r"tasks", views.TaskResultViewSet)
+router.register(r"request-logs", views.RequestLogViewSet, basename="requestlog")
+router.register(r"ping", views.PingViewSet, basename="ping")
+router.register(r"traceroute", views.TraceRouteViewSet, basename="traceroute")
+router.register(r"nmap", views.NmapViewSet, basename="nmap")

netopsio/core/api/views.py

@@ -0,0 +1,92 @@
+"""Netops.io core api views."""
+
+from django_celery_results.models import TaskResult
+from drf_yasg import openapi
+from drf_yasg.utils import swagger_auto_schema
+from rest_framework import viewsets
+from rest_framework.response import Response
+
+from core import models, tasks
+from core.api import serializers
+from netopsio.utilities import get_ip_address
+
+
+class TaskResultViewSet(viewsets.ReadOnlyModelViewSet):
+    """Rest API View for 'list' and 'retrieving' TaskResult actions."""
+
+    queryset = TaskResult.objects.all()
+    serializer_class = serializers.TaskResultSerializer
+    lookup_field = "task_id"
+
+
+class RequestLogViewSet(viewsets.ReadOnlyModelViewSet):
+    """Rest API View for 'list' and 'retrieving' RequestLog actions."""
+
+    queryset = models.RequestLog.objects.all()
+    serializer_class = serializers.RequestLogSerializer
+
+
+class PingViewSet(viewsets.ViewSet):
+    """Ping Viewset."""
+
+    @swagger_auto_schema(
+        request_body=openapi.Schema(
+            type=openapi.TYPE_OBJECT,
+            properties={
+                "host": openapi.Schema(
+                    type=openapi.TYPE_STRING, description="Host to ping."
+                ),
+            },
+        )
+    )  # pylint: disable=no-self-use
+    def create(self, request, host=None):
+        """Ping a host."""
+        host = get_ip_address(request)
+        task = tasks.ping.delay(host=host)
+        data = models.TaskModel(task_id=task.task_id)
+        serializer = serializers.TaskSerializer(data, context={"request": request})
+        return Response(serializer.data)
+
+
+class TraceRouteViewSet(viewsets.ViewSet):
+    """TraceRoute Viewset."""
+
+    @swagger_auto_schema(
+        request_body=openapi.Schema(
+            type=openapi.TYPE_OBJECT,
+            properties={
+                "host": openapi.Schema(
+                    type=openapi.TYPE_STRING, description="Host to traceroute."
+                ),
+            },
+        )
+    )  # pylint: disable=no-self-use
+    def create(self, request, host=None):
+        """Traceroute a host."""
+        host = get_ip_address(request)
+        task = tasks.traceroute.delay(host=host)
+        data = models.TaskModel(task_id=task.task_id)
+        serializer = serializers.TaskSerializer(data, context={"request": request})
+        return Response(serializer.data)
+
+
+class NmapViewSet(viewsets.ViewSet):
+    """Nmap Viewset."""
+
+    @swagger_auto_schema(
+        request_body=openapi.Schema(
+            type=openapi.TYPE_OBJECT,
+            properties={
+                "host": openapi.Schema(
+                    type=openapi.TYPE_STRING, description="Host to ping."
+                ),
+            },
+        )
+    )  # pylint: disable=no-self-use
+    def create(self, request, host=None):
+        """Nmap a host."""
+        host = get_ip_address(request)
+        task = tasks.nmap.delay(host=host)
+        data = models.TaskModel(task_id=task.task_id)
+        serializer = serializers.TaskSerializer(data, context={"request": request})
+        return Response(serializer.data)

netopsio/home/apps.py → netopsio/core/apps.py

@@ -3,7 +3,7 @@
 from django.apps import AppConfig
 
 
-class HomeConfig(AppConfig):
-    "Home app config."
+class CoreConfig(AppConfig):
+    "Core app config."
     default_auto_field = "django.db.models.BigAutoField"
-    name = "home"
+    name = "core"

netopsio/traceroute/migrations/0001_initial.py → netopsio/core/migrations/0001_add_request_log.py

@@ -1,4 +1,4 @@
-# Generated by Django 3.2.10 on 2022-07-11 00:25
+# Generated by Django 3.2.10 on 2022-07-18 02:40
 
 from django.db import migrations, models
 
@@ -11,7 +11,7 @@ class Migration(migrations.Migration):
 
     operations = [
         migrations.CreateModel(
-            name="TraceRouteRequest",
+            name="RequestLog",
             fields=[
                 (
                     "id",
@@ -26,6 +26,7 @@ class Migration(migrations.Migration):
                 ("task_id", models.UUIDField(null=True, unique=True)),
                 ("ip", models.CharField(max_length=255)),
                 ("result", models.TextField()),
+                ("app", models.CharField(max_length=255)),
             ],
         ),
     ]

netopsio/nmap/models.py → netopsio/core/models.py

@@ -1,19 +1,20 @@
-"""Nmap App Models."""
+"""Netops.io core app models."""
 
 from django.db import models
 
 
-class NmapRequest(models.Model):
-    """Nmap Requests Logging Model."""
+class RequestLog(models.Model):
+    """Request Logging Model."""
 
     date = models.DateTimeField(auto_now_add=True)
     task_id = models.UUIDField(unique=True, null=True)
     ip = models.CharField(max_length=255)
     result = models.TextField()
+    app = models.CharField(max_length=255)
 
 
-class Nmap:
-    """Nmap Model."""
+class TaskModel:
+    """Task Model."""
 
     def __init__(self, task_id: str):
         """Initialize the model."""

netopsio/core/tasks.py

@@ -0,0 +1,35 @@
+"""Celery Background Tasks"""
+import subprocess
+
+from celery import shared_task
+
+from core.models import RequestLog
+
+
+@shared_task
+def ping(host: str, count: int = 2) -> str:
+    """Ping host worker."""
+    task = subprocess.run(
+        ["ping", f"-c {count}", host], capture_output=True, check=True
+    )
+    data = RequestLog(ip=host, result=task.stdout.decode("utf-8"), app="ping")
+    data.save()
+    return task.stdout.decode("utf-8")
+
+
+@shared_task
+def traceroute(host: str) -> str:
+    """Traceroute host worker."""
+    task = subprocess.run(["traceroute", host], capture_output=True, check=True)
+    data = RequestLog(ip=host, result=task.stdout.decode("utf-8"), app="traceroute")
+    data.save()
+    return task.stdout.decode("utf-8")
+
+
+@shared_task
+def nmap(host: str) -> str:
+    """Nmap host worker."""
+    task = subprocess.run(["nmap", "-v", "-A", host], capture_output=True, check=True)
+    data = RequestLog(ip=host, result=task.stdout.decode("utf-8"), app="nmap")
+    data.save()
+    return task.stdout.decode("utf-8")

netopsio/nmap/templates/nmap/base.html → netopsio/core/templates/core/task.html

@@ -14,4 +14,4 @@
         </tbody>
     </table>
 </div>
-{% endblock %}
+{% endblock %}

netopsio/core/tests/__init__.py

@@ -0,0 +1 @@
+"""Core tests."""

netopsio/core/tests/test_api.py

@@ -0,0 +1,43 @@
+"""Core api tests."""
+from django.test import TestCase
+from rest_framework.test import APIRequestFactory
+
+from core.api import views
+
+
+class CoreApiTestCase(TestCase):
+    def setUp(self):
+        self.factory = APIRequestFactory()
+        self.ping = views.PingViewSet()
+        self.traceroute = views.TraceRouteViewSet()
+        self.nmap = views.NmapViewSet()
+        self.host = {"host": "localhost"}
+        self.content_type = {"Content-type": "application/json"}
+        return super().setUp()
+
+    def test_ping_api(self):
+        request = self.factory.post(
+            "/api/v1/ping/",
+            self.host,
+            content_type=self.content_type,
+        )
+        response = self.ping.create(request)
+        self.assertEqual(response.status_code, 200)
+
+    def test_traceroute_api(self):
+        request = self.factory.post(
+            "/api/v1/traceroute/",
+            self.host,
+            content_type=self.content_type,
+        )
+        response = self.traceroute.create(request)
+        self.assertEqual(response.status_code, 200)
+
+    def test_nmap_api(self):
+        request = self.factory.post(
+            "/api/v1/nmap/",
+            self.host,
+            content_type=self.content_type,
+        )
+        response = self.nmap.create(request)
+        self.assertEqual(response.status_code, 200)

netopsio/core/tests/test_models.py

@@ -0,0 +1,18 @@
+"Core app tests."
+from django.test import TestCase
+
+from core.models import RequestLog
+
+
+class RequestLogTestCase(TestCase):
+    """RequestLog TestCases."""
+
+    def setUp(self):
+        """Setup tests."""
+        RequestLog.objects.create(ip="1.1.1.1", result="OK", app="ping")
+        return super().setUp()
+
+    def test_requestlog_models(self):
+        """Test RequestLog models."""
+        req = RequestLog.objects.get(ip="1.1.1.1")
+        self.assertEqual(req.ip, "1.1.1.1")

netopsio/core/tests/test_tasks.py

@@ -0,0 +1,23 @@
+"""Core tasks tests."""
+
+
+from django.test import TestCase
+
+from core import tasks
+
+
+class TasksTestCase(TestCase):
+    def setUp(self):
+        self.host = "localhost"
+
+    def test_ping_task(self):
+        ping = tasks.ping.delay(host=self.host)
+        self.assertIn(ping.status, ("PENDING", "STARTED", "SUCCESS"))
+
+    def test_traceroute_task(self):
+        traceroute = tasks.ping.delay(host=self.host)
+        self.assertIn(traceroute.status, ("PENDING", "STARTED", "SUCCESS"))
+
+    def test_nmap_task(self):
+        nmap = tasks.nmap.delay(host=self.host)
+        self.assertIn(nmap.status, ("PENDING", "STARTED", "SUCCESS"))

netopsio/core/tests/test_views.py

@@ -0,0 +1,12 @@
+"Core app tests."
+from django.test import TestCase
+
+
+class CoreViewsTest(TestCase):
+    """Test the Core App views."""
+
+    def test_index_view(self):
+        """Test the core page."""
+        response = self.client.get("/")
+        self.assertTemplateUsed(response, "core/base.html")
+        self.assertEqual(response.status_code, 200)

netopsio/core/urls.py

@@ -0,0 +1,13 @@
+"Core app urls."
+from django.urls import path
+
+from core import views
+
+urlpatterns = [
+    path("", views.index, name="core-index"),
+    path("tasks/", views.tasks_view, name="tasks"),
+    path("tasks/<str:task_id>/", views.task_details, name="task-details"),
+    path("ping/", views.ping, name="ping"),
+    path("traceroute/", views.traceroute, name="traceroute"),
+    path("nmap/", views.nmap, name="nmap"),
+]