Update test dependencies

CHANGELOG.rst

@@ -1,6 +1,11 @@
 [UNRELEASED] - Under development
 ********************************
 
+Changed
+=======
+- Updated python environment installation from 3.9 to 3.11
+- Updated test dependencies
+
 [2023.2.0] - 2024-02-16
 ***********************
 

controllers/__init__.py

@@ -1,4 +1,5 @@
 """PipelineController"""
+
 # pylint: disable=unnecessary-lambda,invalid-name,unnecessary-comprehension
 import os
 from datetime import datetime
@@ -51,7 +52,7 @@ def insert_pipeline(self, pipeline: Dict) -> InsertOneResult:
                         "inserted_at": utc_now,
                         "updated_at": utc_now,
                     }
-                ).dict(exclude_none=True)
+                ).model_dump(exclude_none=True)
             )
         except ValidationError as err:
             raise err

db/models.py

@@ -1,21 +1,23 @@
 """DB Models"""
+
 # pylint: disable=no-self-argument,invalid-name,no-name-in-module
 from datetime import datetime
 from typing import Dict, List, Optional, Union
 
-from pydantic import BaseModel, Field, conint, root_validator, validator
+from pydantic import BaseModel, Field, field_validator, model_validator
+from typing_extensions import Annotated
 
 
 class DocumentBaseModel(BaseModel):
     """Base model for Mongo documents"""
 
     id: str = Field(None, alias="_id")
-    inserted_at: Optional[datetime]
-    updated_at: Optional[datetime]
+    inserted_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
 
-    def dict(self, **kwargs) -> Dict:
+    def model_dump(self, **kwargs) -> Dict:
         """Return a dictionary representation of the model"""
-        values = super().dict(**kwargs)
+        values = super().model_dump(**kwargs)
         if "id" in values and values["id"]:
             values["_id"] = values["id"]
         if "exclude" in kwargs and "_id" in kwargs["exclude"]:
@@ -26,48 +28,49 @@ def dict(self, **kwargs) -> Dict:
 class MatchSubDoc(BaseModel):
     """Match DB SubDocument Model."""
 
-    in_port: Optional[int]
-    dl_src: Optional[str]
-    dl_dst: Optional[str]
-    dl_type: Optional[int]
-    dl_vlan: Optional[Union[int, str]]
-    dl_vlan_pcp: Optional[int]
-    nw_src: Optional[str]
-    nw_dst: Optional[str]
-    nw_proto: Optional[int]
-    tp_src: Optional[int]
-    tp_dst: Optional[int]
-    in_phy_port: Optional[int]
-    ip_dscp: Optional[int]
-    ip_ecn: Optional[int]
-    udp_src: Optional[int]
-    udp_dst: Optional[int]
-    sctp_src: Optional[int]
-    sctp_dst: Optional[int]
-    icmpv4_type: Optional[int]
-    icmpv4_code: Optional[int]
-    arp_op: Optional[int]
-    arp_spa: Optional[str]
-    arp_tpa: Optional[str]
-    arp_sha: Optional[str]
-    arp_tha: Optional[str]
-    ipv6_src: Optional[str]
-    ipv6_dst: Optional[str]
-    ipv6_flabel: Optional[int]
-    icmpv6_type: Optional[int]
-    icmpv6_code: Optional[int]
-    nd_tar: Optional[int]
-    nd_sll: Optional[int]
-    nd_tll: Optional[int]
-    mpls_lab: Optional[int]
-    mpls_tc: Optional[int]
-    mpls_bos: Optional[int]
-    pbb_isid: Optional[int]
-    v6_hdr: Optional[int]
-    metadata: Optional[int]
-    tun_id: Optional[int]
-
-    @validator("dl_vlan")
+    in_port: Optional[int] = None
+    dl_src: Optional[str] = None
+    dl_dst: Optional[str] = None
+    dl_type: Optional[int] = None
+    dl_vlan: Optional[Union[int, str]] = None
+    dl_vlan_pcp: Optional[int] = None
+    nw_src: Optional[str] = None
+    nw_dst: Optional[str] = None
+    nw_proto: Optional[int] = None
+    tp_src: Optional[int] = None
+    tp_dst: Optional[int] = None
+    in_phy_port: Optional[int] = None
+    ip_dscp: Optional[int] = None
+    ip_ecn: Optional[int] = None
+    udp_src: Optional[int] = None
+    udp_dst: Optional[int] = None
+    sctp_src: Optional[int] = None
+    sctp_dst: Optional[int] = None
+    icmpv4_type: Optional[int] = None
+    icmpv4_code: Optional[int] = None
+    arp_op: Optional[int] = None
+    arp_spa: Optional[str] = None
+    arp_tpa: Optional[str] = None
+    arp_sha: Optional[str] = None
+    arp_tha: Optional[str] = None
+    ipv6_src: Optional[str] = None
+    ipv6_dst: Optional[str] = None
+    ipv6_flabel: Optional[int] = None
+    icmpv6_type: Optional[int] = None
+    icmpv6_code: Optional[int] = None
+    nd_tar: Optional[int] = None
+    nd_sll: Optional[int] = None
+    nd_tll: Optional[int] = None
+    mpls_lab: Optional[int] = None
+    mpls_tc: Optional[int] = None
+    mpls_bos: Optional[int] = None
+    pbb_isid: Optional[int] = None
+    v6_hdr: Optional[int] = None
+    metadata: Optional[int] = None
+    tun_id: Optional[int] = None
+
+    @field_validator("dl_vlan")
+    @classmethod
     def vlan_with_mask(cls, v):
         """Validate vlan format"""
         try:
@@ -86,26 +89,26 @@ class TableMissDoc(BaseModel):
     """Base model for Table miss flow"""
 
     priority: int
-    instructions: Optional[List[dict]]
-    match: Optional[MatchSubDoc]
+    instructions: Optional[List[dict]] = None
+    match: Optional[MatchSubDoc] = None
 
 
 class MultitableDoc(BaseModel):
     """Base model for Multitable"""
 
-    table_id: conint(ge=0, le=254)
-    table_miss_flow: Optional[TableMissDoc]
-    description: Optional[str]
-    napps_table_groups: Optional[dict[str, List[str]]]
+    table_id: Annotated[int, Field(ge=0, le=254)]
+    table_miss_flow: Optional[TableMissDoc] = None
+    description: Optional[str] = None
+    napps_table_groups: Optional[dict[str, List[str]]] = None
 
-    @root_validator
-    def validate_intructions(cls, values):
+    @model_validator(mode="after")
+    def validate_intructions(self):
         """Validate intructions"""
-        table_miss_flow = values.get("table_miss_flow")
+        table_miss_flow = self.table_miss_flow
         if not table_miss_flow:
-            return values
-        table_id = values["table_id"]
-        instructions = table_miss_flow.dict(exclude_none=True)["instructions"]
+            return self
+        table_id = self.table_id
+        instructions = table_miss_flow.model_dump(exclude_none=True)["instructions"]
         for instruction in instructions:
             miss_table_id = instruction.get("table_id")
             if miss_table_id is not None and miss_table_id <= table_id:
@@ -114,22 +117,23 @@ def validate_intructions(cls, values):
                     f"table_id {miss_table_id} in instructions"
                 )
                 raise ValueError(msg)
-        return values
+        return self
 
 
 class PipelineBaseDoc(DocumentBaseModel):
     """Base model for Pipeline documents"""
 
-    status = "disabled"
+    status: str = "disabled"
     multi_table: List[MultitableDoc]
 
-    @validator("multi_table")
+    @field_validator("multi_table")
+    @classmethod
     def validate_table_groups(cls, pipeline):
         """Validate table groups"""
         content = {}
         id_set = set()
         for table in pipeline:
-            table_dict = table.dict(exclude_none=True)
+            table_dict = table.model_dump(exclude_none=True)
             table_groups = table_dict.get("napps_table_groups", {})
             table_id = table_dict["table_id"]
             if table_id in id_set:

main.py

@@ -2,6 +2,7 @@
 
 This NApp implements Oplenflow multi tables
 """
+
 # pylint: disable=unused-argument, too-many-arguments, too-many-public-methods
 # pylint: disable=attribute-defined-outside-init
 import pathlib

requirements/run.in

@@ -1 +1 @@
-requests==2.27.0
+requests==2.31.0

requirements/run.txt

@@ -1,16 +1,16 @@
 #
-# This file is autogenerated by pip-compile with Python 3.9
+# This file is autogenerated by pip-compile with Python 3.11
 # by the following command:
 #
 #    pip-compile --output-file=requirements/run.txt requirements/run.in
 #
-certifi==2021.10.8
+certifi==2024.2.2
     # via requests
-charset-normalizer==2.0.10
+charset-normalizer==3.3.2
     # via requests
-idna==3.3
+idna==3.6
     # via requests
-requests==2.27.0
+requests==2.31.0
     # via -r requirements/run.in
-urllib3==1.26.7
+urllib3==1.26.18
     # via requests

settings.py

@@ -1,4 +1,5 @@
 """Module with the Constants used in the kytos/of_multi_table."""
+
 FLOW_MANAGER_URL = "http://localhost:8181/api/kytos/flow_manager"
 COOKIE_PREFIX = 0xAD
 

setup.py

@@ -3,6 +3,7 @@
 Run "python3 setup.py --help-commands" to list all available commands and their
 descriptions.
 """
+
 import json
 import os
 import shutil
@@ -276,7 +277,7 @@ def read_requirements(path="requirements/run.txt"):
     classifiers=[
         "License :: OSI Approved :: MIT License",
         "Operating System :: POSIX :: Linux",
-        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.11",
         "Topic :: System :: Networking",
     ],
 )

status.py

@@ -1,4 +1,5 @@
 """Defined status for pipeline"""
+
 from enum import Enum
 
 

tests/unit/test_controllers.py

@@ -1,4 +1,5 @@
 """Test the Pipeline controllers"""
+
 from unittest.mock import MagicMock
 
 import pytest

tests/unit/test_db_models.py

@@ -1,4 +1,5 @@
 """Tests for DB models"""
+
 import pytest
 from db.models import PipelineBaseDoc
 from pydantic import ValidationError

tests/unit/test_main.py

@@ -1,8 +1,10 @@
 """Test the Main class"""
+
+import asyncio
 from unittest.mock import MagicMock, patch
 
 from napps.kytos.of_multi_table.main import Main
-from pydantic import BaseModel, ValidationError
+from pydantic import ValidationError
 
 from kytos.lib.helpers import get_controller_mock, get_test_client
 
@@ -475,9 +477,9 @@ async def test_send_flows(self, _):
         self.napp.send_flows(flows, "install", True)
         assert self.napp.controller.buffers.app.put.call_count == 2
 
-    async def test_add_pipeline(self, event_loop):
+    async def test_add_pipeline(self):
         """Test adding a pipeline"""
-        self.napp.controller.loop = event_loop
+        self.napp.controller.loop = asyncio.get_running_loop()
         payload = {
             "status": "disabled",
             "multi_table": [
@@ -506,20 +508,22 @@ async def test_add_pipeline(self, event_loop):
         response = await api.post(url, json=payload)
         assert response.status_code == 201
 
-    async def test_add_pipeline_error_empty_json(self, event_loop):
+    async def test_add_pipeline_error_empty_json(self):
         """Test adding pipeline with an empty JSON"""
-        self.napp.controller.loop = event_loop
+        self.napp.controller.loop = asyncio.get_running_loop()
         payload = {"multi_table": [{"table_id": 299}]}
         controller = self.napp.pipeline_controller
-        controller.insert_pipeline.side_effect = ValidationError("", BaseModel)
+        controller.insert_pipeline.side_effect = ValidationError.from_exception_data(
+            "", []
+        )
         api = get_test_client(self.napp.controller, self.napp)
         url = f"{self.base_endpoint}/pipeline"
         response = await api.post(url, json=payload)
         assert response.status_code == 400
 
-    async def test_add_pipeline_error_empty_content(self, event_loop):
+    async def test_add_pipeline_error_empty_content(self):
         """Test adding pipeline with no content"""
-        self.napp.controller.loop = event_loop
+        self.napp.controller.loop = asyncio.get_running_loop()
         api = get_test_client(self.napp.controller, self.napp)
         url = f"{self.base_endpoint}/pipeline"
         response = await api.post(url)

tox.ini

@@ -4,19 +4,19 @@ envlist = coverage,lint
 
 [testenv]
 whitelist_externals = rm
-deps = -Urrequirements/dev.in
+deps = -rrequirements/dev.in
 setenv=
-    PYTHONPATH = {toxworkdir}/py39/var/lib/kytos/:{envdir}
+    PYTHONPATH = {toxworkdir}/py311/var/lib/kytos/:{envdir}
 
 
 [testenv:coverage]
 skip_install = true
-envdir = {toxworkdir}/py39
+envdir = {toxworkdir}/py311
 commands=
     python3 setup.py coverage {posargs}
 
 
 [testenv:lint]
 skip_install = true
-envdir = {toxworkdir}/py39
+envdir = {toxworkdir}/py311
 commands = python3 setup.py lint