oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#1943: unit tests + refactoring HTTP -> Http

Browse Source
This commit is contained in:
Czar.Echavez
2023-10-06 12:10:57 +01:00
parent be6b904db9
commit 853bb9eecc
9 changed files with 147 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/primaite/simulator/system/applications/web_browser.py
View File

@@ -2,7 +2,7 @@ from ipaddress import IPv4Address
from typing import Dict, Optional
from urllib.parse import urlparse
from primaite.simulator.network.protocols.http import HTTPRequestMethod, HTTPRequestPacket, HTTPResponsePacket
from primaite.simulator.network.protocols.http import HttpRequestMethod, HttpRequestPacket, HttpResponsePacket
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.applications.application import Application
@@ -19,7 +19,7 @@ class WebBrowser(Application):
domain_name_ip_address: Optional[IPv4Address] = None
"The IP address of the domain name for the webpage."
latest_response: HTTPResponsePacket = None
latest_response: HttpResponsePacket = None
"""Keeps track of the latest HTTP response."""
def __init__(self, **kwargs):
@@ -87,7 +87,7 @@ class WebBrowser(Application):
return False
# create HTTPRequest payload
payload = HTTPRequestPacket(request_method=HTTPRequestMethod.GET, request_url=url)
payload = HttpRequestPacket(request_method=HttpRequestMethod.GET, request_url=url)
# send request
return self.send(
@@ -98,7 +98,7 @@ class WebBrowser(Application):
def send(
self,
payload: HTTPRequestPacket,
payload: HttpRequestPacket,
dest_ip_address: Optional[IPv4Address] = None,
dest_port: Optional[Port] = Port.HTTP,
session_id: Optional[str] = None,
@@ -120,7 +120,7 @@ class WebBrowser(Application):
payload=payload, dest_ip_address=dest_ip_address, dest_port=dest_port, session_id=session_id, **kwargs
)
def receive(self, payload: HTTPResponsePacket, session_id: Optional[str] = None, **kwargs) -> bool:
def receive(self, payload: HttpResponsePacket, session_id: Optional[str] = None, **kwargs) -> bool:
"""
Receives a payload from the SessionManager.
@@ -128,8 +128,8 @@ class WebBrowser(Application):
:param session_id: The Session ID the payload is to originate from. Optional.
:return: True if successful, False otherwise.
"""
if not isinstance(payload, HTTPResponsePacket):
self.sys_log.error(f"{self.name} received a packet that is not an HTTPResponsePacket")
if not isinstance(payload, HttpResponsePacket):
self.sys_log.error(f"{self.name} received a packet that is not an HttpResponsePacket")
return False
self.sys_log.info(f"{self.name}: Received HTTP {payload.status_code.value}")
self.latest_response = payload

42
src/primaite/simulator/system/services/web_server/web_server_service.py → src/primaite/simulator/system/services/web_server/web_server.py
View File

@@ -3,10 +3,10 @@ from typing import Any, Optional
from urllib.parse import urlparse
from primaite.simulator.network.protocols.http import (
HTTPRequestMethod,
HTTPRequestPacket,
HTTPResponsePacket,
HTTPStatusCode,
HttpRequestMethod,
HttpRequestPacket,
HttpResponsePacket,
HttpStatusCode,
)
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
@@ -37,52 +37,52 @@ class WebServer(Service):
# index HTML main file
self.file_system.create_file(file_name="index.html", folder_name="primaite", real=True)
def _process_http_request(self, payload: HTTPRequestPacket, session_id: Optional[str] = None) -> bool:
def _process_http_request(self, payload: HttpRequestPacket, session_id: Optional[str] = None) -> bool:
"""
Parse the HTTPRequestPacket.
Parse the HttpRequestPacket.
:param: payload: Payload containing th HTTPRequestPacket
:type: payload: HTTPRequestPacket
:param: payload: Payload containing th HttpRequestPacket
:type: payload: HttpRequestPacket
:param: session_id: Session id of the http request
:type: session_id: Optional[str]
"""
response = HTTPResponsePacket()
response = HttpResponsePacket()
self.sys_log.info(f"{self.name}: Received HTTP {payload.request_method.name} {payload.request_url}")
# check the type of HTTP request
if payload.request_method == HTTPRequestMethod.GET:
if payload.request_method == HttpRequestMethod.GET:
response = self._handle_get_request(payload=payload)
elif payload.request_method == HTTPRequestMethod.POST:
elif payload.request_method == HttpRequestMethod.POST:
pass
else:
# send a method not allowed response
response.status_code = HTTPStatusCode.METHOD_NOT_ALLOWED
response.status_code = HttpStatusCode.METHOD_NOT_ALLOWED
# send response to web client
self.send(payload=response, session_id=session_id)
# return true if response is OK
return response.status_code == HTTPStatusCode.OK
return response.status_code == HttpStatusCode.OK
def _handle_get_request(self, payload: HTTPRequestPacket) -> HTTPResponsePacket:
def _handle_get_request(self, payload: HttpRequestPacket) -> HttpResponsePacket:
"""
Handle a GET HTTP request.
:param: payload: HTTP request payload
:type: payload: HTTPRequestPacket
:type: payload: HttpRequestPacket
"""
response = HTTPResponsePacket(status_code=HTTPStatusCode.BAD_REQUEST, payload=payload)
response = HttpResponsePacket(status_code=HttpStatusCode.NOT_FOUND, payload=payload)
try:
parsed_url = urlparse(payload.request_url)
path = parsed_url.path.strip("/")
if len(path) < 1:
# query succeeded
response.status_code = HTTPStatusCode.OK
response.status_code = HttpStatusCode.OK
if path.startswith("users"):
# get data from DatabaseServer
@@ -90,17 +90,17 @@ class WebServer(Service):
# get all users
if db_client.query("SELECT * FROM user;"):
# query succeeded
response.status_code = HTTPStatusCode.OK
response.status_code = HttpStatusCode.OK
return response
except Exception:
# something went wrong on the server
response.status_code = HTTPStatusCode.INTERNAL_SERVER_ERROR
response.status_code = HttpStatusCode.INTERNAL_SERVER_ERROR
return response
def send(
self,
payload: HTTPResponsePacket,
payload: HttpResponsePacket,
session_id: Optional[str] = None,
dest_ip_address: Optional[IPv4Address] = None,
dest_port: Optional[Port] = None,
@@ -138,7 +138,7 @@ class WebServer(Service):
:param: session_id: The id of the session. Optional.
"""
# check if the payload is an HTTPPacket
if not isinstance(payload, HTTPRequestPacket):
if not isinstance(payload, HttpRequestPacket):
self.sys_log.error("Payload is not an HTTPPacket")
return False