chore: Upgraded python and dependencies

.github/workflows/pythonpackage.yml

@@ -13,7 +13,7 @@ jobs:
  - name: Set up Python
  uses: actions/setup-python@v1
  with:
- python-version: '3.9'
+ python-version: '3.11'
  - name: Install dependencies
  run: |
  python -m pip install --upgrade pip

.github/workflows/unittest.yml

@@ -13,7 +13,7 @@ jobs:
  runs-on: ubuntu-latest
  strategy:
  matrix:
- python-version: [3.9]
+ python-version: [3.11]
 
  steps:
  - uses: actions/checkout@v1

.scrutinizer.yml

@@ -4,7 +4,7 @@ checks:
  duplicate_code: true
 build:
  environment:
- python: 3.9.12
+ python: 3.11.8
  postgresql: false
  redis: false
  dependencies:

.travis.yml

@@ -4,6 +4,7 @@ python:
  - "3.7"
  - "3.8"
  - "3.9"
+ - "3.11"
 install:
  - pip install --upgrade pip
  - pip install -r requirements/dev.txt

kytos/core/api_server.py

@@ -404,8 +404,7 @@ def _get_decorated_functions(napp):
  and isinstance(pub_attr.route_index, int)
  ):
  callables.append(pub_attr)
- for pub_attr in sorted(callables, key=lambda f: f.route_index):
- yield pub_attr
+ yield from sorted(callables, key=lambda f: f.route_index)
 
  @classmethod
  def get_absolute_rule(cls, rule, napp):

kytos/core/atcp_server.py

@@ -116,7 +116,7 @@ def __init__(self):
  if not self.server:
  raise ValueError("server instance must be assigned before init")
 
- def connection_made(self, transport):
+ def connection_made(self, transport: asyncio.Transport):
  """Handle new client connection, passing it to the controller.
 
  Build a new Kytos `Connection` and send a ``kytos/core.connection.new``
@@ -130,7 +130,7 @@ def connection_made(self, transport):
 
  LOG.info("New connection from %s:%s", addr, port)
 
- self.connection = Connection(addr, port, socket)
+ self.connection = Connection(addr, port, socket, transport)
 
  # This allows someone to inherit from KytosServer and start a server
  # on another port to handle a different protocol.

kytos/core/auth.py

@@ -25,7 +25,7 @@
  content_type_json_or_415, error_msg,
  get_json_or_400)
 from kytos.core.retry import before_sleep, for_all_methods, retries
-from kytos.core.user import HashSubDoc, UserDoc, UserDocUpdate
+from kytos.core.user import HashSubDoc, UserDoc, UserDocUpdate, hashing
 
 __all__ = ['authenticated']
 
@@ -110,7 +110,7 @@ def create_user(self, user_data: dict) -> InsertOneResult:
  "email": user_data.get('email'),
  "inserted_at": utc_now,
  "updated_at": utc_now,
- }).dict())
+ }).model_dump())
  except DuplicateKeyError as err:
  raise err
  except ValidationError as err:
@@ -142,7 +142,7 @@ def update_user(self, username: str, data: dict) -> dict:
  "$set": UserDocUpdate(**{
  **data,
  **{"updated_at": utc_now}
- }).dict(exclude_none=True)
+ }).model_dump(exclude_none=True)
  },
  return_document=ReturnDocument.AFTER
  )
@@ -307,7 +307,7 @@ def _authenticate_user(self, request: Request) -> JSONResponse:
  user = self._find_user(username)
  if user["state"] != 'active':
  raise HTTPException(401, detail='This user is not active')
- password_hashed = UserDoc.hashing(password, user["hash"])
+ password_hashed = hashing(password, user["hash"])
  if user["password"] != password_hashed:
  raise HTTPException(401, detail="Incorrect password")
  time_exp = datetime.datetime.utcnow() + datetime.timedelta(

kytos/core/connection.py

@@ -1,9 +1,9 @@
 """Module with main classes related to Connections."""
 import logging
+from asyncio import Transport, trsock
 from enum import Enum
 from errno import EBADF, ENOTCONN
 from socket import SHUT_RDWR
-from socket import error as SocketError
 
 __all__ = ('Connection', 'ConnectionProtocol', 'ConnectionState')
 
@@ -33,19 +33,28 @@ def __init__(self, name=None, version=None, state=None):
 class Connection:
  """Connection class to abstract a network connections."""
 
- def __init__(self, address, port, socket, switch=None):
+ def __init__(
+ self,
+ address: str,
+ port: int,
+ socket: trsock.TransportSocket,
+ transport: Transport,
+ switch=None
+ ):
  """Assign parameters to instance variables.
 
  Args:
  address (|hw_address|): Source address.
  port (int): Port number.
- socket (socket): socket.
+ socket (TransportSocket): socket.
+ transport (Transport): transport.
  switch (:class:`~.Switch`): switch with this connection.
  """
  self.address = address
  self.port = port
  self.socket = socket
  self.switch = switch
+ self.transport = transport
  self.state = ConnectionState.NEW
  self.protocol = ConnectionProtocol()
  self.remaining_data = b''
@@ -65,6 +74,7 @@ def state(self):
  @state.setter
  def state(self, new_state):
  if new_state not in ConnectionState:
+ # pylint: disable=broad-exception-raised
  raise Exception('Unknown State', new_state)
  # pylint: disable=attribute-defined-outside-init
  self._state = new_state
@@ -90,8 +100,8 @@ def send(self, buffer):
  """
  try:
  if self.is_alive():
- self.socket.sendall(buffer)
- except (OSError, SocketError) as exception:
+ self.transport.write(buffer)
+ except (OSError, TypeError) as exception:
  LOG.debug('Could not send packet. Exception: %s', exception)
  self.close()
  raise
@@ -103,7 +113,7 @@ def close(self):
 
  try:
  self.socket.shutdown(SHUT_RDWR)
- self.socket.close()
+ self.transport.close()
  self.socket = None
  LOG.debug('Connection Closed: %s', self.id)
  except OSError as exception:

kytos/core/controller.py

@@ -28,7 +28,6 @@
 from importlib import reload as reload_module
 from importlib.util import module_from_spec, spec_from_file_location
 from pathlib import Path
-from socket import error as SocketError
 
 from pyof.foundation.exceptions import PackException
 
@@ -38,7 +37,7 @@
 from kytos.core.auth import Auth
 from kytos.core.buffers import KytosBuffers
 from kytos.core.config import KytosConfig
-from kytos.core.connection import ConnectionState
+from kytos.core.connection import Connection, ConnectionState
 from kytos.core.db import db_conn_wait
 from kytos.core.dead_letter import DeadLetter
 from kytos.core.events import KytosEvent
@@ -107,7 +106,7 @@ def __init__(self, options=None, loop: AbstractEventLoop = None):
  #: This dict stores all connections between the controller and the
  #: switches. The key for this dict is a tuple (ip, port). The content
  #: is a Connection
- self.connections = {}
+ self.connections: dict[tuple, Connection] = {}
  #: dict: mapping of events and event listeners.
  #:
  #: The key of the dict is a KytosEvent (or a string that represent a
@@ -155,6 +154,9 @@ def __init__(self, options=None, loop: AbstractEventLoop = None):
  self._alisten_tasks = set()
  self.qmonitors: list[QueueMonitorWindow] = []
 
+ #: APM client in memory to be closed when necessary
+ self.apm = None
+
  self._register_endpoints()
  #: Adding the napps 'enabled' directory into the PATH
  #: Now you can access the enabled napps with:
@@ -259,7 +261,7 @@ def start(self, restart=False):
  db_conn_wait(db_backend=self.options.database)
  self.start_auth()
  if self.options.apm:
- init_apm(self.options.apm, app=self.api_server.app)
+ self.apm = init_apm(self.options.apm, app=self.api_server.app)
  if not restart:
  self.create_pidfile()
  self.start_controller()
@@ -509,6 +511,9 @@ def stop_controller(self, graceful=True):
  # self.server.server_close()
 
  self.stop_queue_monitors()
+ if self.apm:
+ self.log.info("Stopping APM server...")
+ self.apm.close()
  self.log.info("Stopping API Server...")
  self.api_server.stop()
  self.log.info("Stopped API Server")
@@ -626,7 +631,7 @@ async def msg_out_event_handler(self):
  message.header.xid,
  packet.hex())
  self.notify_listeners(triggered_event)
- except (OSError, SocketError):
+ except (OSError, TypeError):
  await self.publish_connection_error(triggered_event)
  self.log.info("connection closed. Cannot send message")
  except PackException as err:

kytos/core/dead_letter.py

@@ -5,7 +5,7 @@
 from typing import List
 
 # pylint: disable=no-name-in-module,invalid-name
-from pydantic import BaseModel, ValidationError, constr
+from pydantic import BaseModel, Field, ValidationError
 
 # pylint: enable=no-name-in-module
 from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
@@ -22,7 +22,7 @@ class KytosQueueBufferNames(str, Enum):
 class DeadLetterDeletePayload(BaseModel):
  """DeadLetterDeletePayload."""
 
- event_name: constr(min_length=1)
+ event_name: str = Field(min_length=1)
  ids: List[str] = []
 
 

kytos/core/helpers.py

@@ -9,10 +9,8 @@
 from pathlib import Path
 from threading import Thread
 
-from openapi_core.spec import Spec
-from openapi_core.spec.shortcuts import create_spec
-from openapi_core.validation.request import openapi_request_validator
-from openapi_core.validation.request.datatypes import RequestValidationResult
+from openapi_core import Spec, unmarshal_request
+from openapi_core.exceptions import OpenAPIError
 from openapi_spec_validator import validate_spec
 from openapi_spec_validator.readers import read_from_filename
 
@@ -181,7 +179,7 @@ def handler_context_apm(*args, apm_client=None):
  handler_func, kwargs = handler_context, {}
  if get_apm_name() == "es":
  handler_func = handler_context_apm
- kwargs = dict(apm_client=ElasticAPM.get_client())
+ kwargs = {"apm_client": ElasticAPM.get_client()}
 
  def get_executor(pool, event, default_pool="app", handler=handler):
  """Get executor."""
@@ -271,7 +269,7 @@ async def handler_context_apm(*args, apm_client=None):
  handler_func, kwargs = handler_context, {}
  if get_apm_name() == "es":
  handler_func = handler_context_apm
- kwargs = dict(apm_client=ElasticAPM.get_client())
+ kwargs = {"apm_client": ElasticAPM.get_client()}
 
  async def inner(*args):
  """Inner decorated with events attribute."""
@@ -359,36 +357,22 @@ def _read_from_filename(yml_file_path: Path) -> dict:
 
 def load_spec(yml_file_path: Path):
  """Load and validate spec object given a yml file path."""
- spec_dict = _read_from_filename(yml_file_path)
- validate_spec(spec_dict)
- return create_spec(spec_dict)
-
-
-def _request_validation_result_or_400(result: RequestValidationResult) -> None:
- """Request validation result or raise HTTP 400."""
- if not result.errors:
- return
- error_response = (
- "The request body contains invalid API data."
- )
- errors = result.errors[0]
- if hasattr(errors, "schema_errors"):
- schema_errors = errors.schema_errors[0]
- error_log = {
- "error_message": schema_errors.message,
- "error_validator": schema_errors.validator,
- "error_validator_value": schema_errors.validator_value,
- "error_path": list(schema_errors.path),
- "error_schema": schema_errors.schema,
- "error_schema_path": list(schema_errors.schema_path),
- }
- LOG.debug(f"Invalid request (API schema): {error_log}")
- error_response += f" {schema_errors.message} for field"
- error_response += (
- f" {'/'.join(map(str,schema_errors.path))}."
- )
- else:
+ spec = _read_from_filename(yml_file_path)
+ validate_spec(spec)
+ return Spec.from_dict(spec)
+
+
+def _request_validation_result_or_400(errors: OpenAPIError) -> None:
+ """Raise HTTP 400."""
+ error_response = "The request body contains invalid API data."
+ if not errors.__cause__:
  error_response = str(errors)
+ elif (hasattr(errors.__cause__, "schema_errors") and
+ errors.__cause__.schema_errors):
+ schema_errors = errors.__cause__.schema_errors
+ error_response += f" {schema_errors[0].message}"
+ for error in schema_errors[1:]:
+ error_response += f", {error.message}"
  raise HTTPException(400, detail=error_response)
 
 
@@ -405,8 +389,10 @@ def validate_openapi_request(
  if body:
  content_type_json_or_415(request)
  openapi_request = StarletteOpenAPIRequest(request, body)
- result = openapi_request_validator.validate(spec, openapi_request)
- _request_validation_result_or_400(result)
+ try:
+ unmarshal_request(openapi_request, spec)
+ except OpenAPIError as err:
+ _request_validation_result_or_400(err)
  return body
 
 
@@ -430,8 +416,10 @@ async def avalidate_openapi_request(
  if body:
  content_type_json_or_415(request)
  openapi_request = AStarletteOpenAPIRequest(request, body)
- result = openapi_request_validator.validate(spec, openapi_request)
- _request_validation_result_or_400(result)
+ try:
+ unmarshal_request(openapi_request, spec)
+ except OpenAPIError as err:
+ _request_validation_result_or_400(err)
  return body
 
 

kytos/core/kytosd.py

@@ -77,7 +77,7 @@ def start_shell(controller=None):
 
  # Avoiding sqlite3.ProgrammingError when trying to save command history
  # on Kytos shutdown
- cfg.HistoryAccessor.enabled = False
+ cfg.HistoryAccessor.enabled = True
 
  ipshell = InteractiveShellEmbed(config=cfg,
  banner1=banner1,