kytos-ng / flow_stats

        expected = flow_by_sw

        assert response.json() == expected

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id_without_dpid(self,

                                                           mock_from_flow):

        """Test flow_stats rest call."""

        flow_info = {

            "byte_count": 10,

            "duration_sec": 20,

            "duration_nsec": 30,

            "packet_count": 40,

            "cookie": 12310228866111668291,

            "match": {"in_port": 1},

            "priority": 32768

            }

        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}

        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}

        mock_from_flow.return_value = flow_by_sw

        endpoint = "/flow/stats"

        url = f"{self.base_endpoint}{endpoint}"

        response = await self.api_client.get(url)

        assert response.status_code == 200

        expected = flow_by_sw

        assert response.json() == expected

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):

        expected = flow_by_sw

        assert response.json() == expected

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id_with_dpid(self, mock_from_flow):

        """Test flow_stats rest call."""

        flow_info = {

            "byte_count": 10,

            "duration_sec": 20,

            "duration_nsec": 30,

            "packet_count": 40,

            "cookie": 12310228866111668291,

            "match": {"in_port": 1},

            "priority": 32768

            }

        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}

        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}

        mock_from_flow.return_value = flow_by_sw

        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"

        url = f"{self.base_endpoint}{endpoint}"

        response = await self.api_client.get(url)

        assert response.status_code == 200

        expected = flow_by_sw

        assert response.json() == expected

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id_not_found(self, mock_from_flow):

        assert response.status_code == 200

        assert len(response.json()) == 1

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):

        """Test flow_stats rest call."""

        flow_info = {

            "byte_count": 10,

            "duration_sec": 20,

            "duration_nsec": 30,

            "packet_count": 40,

            "cookie": 12310228866111668291,

            "match": {"in_port": 1},

            "priority": 32768

            }

        flow_stats_dict_mock = {'6055f13593fad45e0b4699f49d56b105': flow_info}

        flow_by_sw = {"00:00:00:00:00:00:00:01": flow_stats_dict_mock}

        mock_from_flow.return_value = flow_by_sw

        endpoint = "/flow/stats?dpid=00:00:00:00:00:00:00:01"

        url = f"{self.base_endpoint}{endpoint}"

        response = await self.api_client.get(url)

        assert response.status_code == 200

        expected = flow_by_sw

        assert response.json() == expected

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id_without_dpid(self,

        assert response.status_code == 200

        assert len(response.json()) == 1

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flows_counters_bytes(self, mock_from_flow):

        """Test flows_counters function for bytes"""

        flow_info = {

            "byte_count": 10,

            "duration_sec": 20,

            "duration_nsec": 30,

            "packet_count": 40,

            "cookie": 12310228866111668291,

            "match": {"in_port": 1},

            "priority": 32768

            }

        flow_id = '6055f13593fad45e0b4699f49d56b105'

        flow_stats_dict_mock = {flow_id: flow_info}

        dpid = "00:00:00:00:00:00:00:01"

        flow_by_sw = {dpid: flow_stats_dict_mock}

        mock_from_flow.return_value = flow_by_sw

        endpoint = f"{self.base_endpoint}/bytes_count/per_flow/{dpid}"

        response = await self.api_client.get(endpoint)

        assert response.status_code == 200

        assert len(response.json()) == 1

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flow_stats_by_dpid_flow_id(self, mock_from_flow):

        assert json_response[0]["bytes_counter"] == 10

        assert json_response[0]["bits_per_second"] == 4.0

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flows_counters_packet(self, mock_from_flow):

        """Test flows_counters function for packet"""

        flow_info = {

            "byte_count": 10,

            "duration_sec": 20,

            "duration_nsec": 30,

            "packet_count": 40,

            "cookie": 12310228866111668291,

            "match": {"in_port": 1},

            "priority": 32768

            }

        flow_id = '6055f13593fad45e0b4699f49d56b105'

        flow_stats_dict_mock = {flow_id: flow_info}

        dpid = "00:00:00:00:00:00:00:01"

        flow_by_sw = {dpid: flow_stats_dict_mock}

        mock_from_flow.return_value = flow_by_sw

        endpoint = f"{self.base_endpoint}/packet_count/per_flow/{dpid}"

        response = await self.api_client.get(endpoint)

        assert response.status_code == 200

        assert len(response.json()) == 1

    @patch("napps.amlight.kytos_stats.main.Main.flow_stats_by_dpid_flow_id")

    async def test_flows_counters_bytes(self, mock_from_flow):