• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2022 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Cross-language pw_transfer tests that are as small/fast as possible.
15
16Usage:
17
18   bazel run pw_transfer/integration_test:cross_language_small_test
19
20Command-line arguments must be provided after a double-dash:
21
22   bazel run pw_transfer/integration_test:cross_language_small_test -- \
23       --server-port 3304
24
25Which tests to run can be specified as command-line arguments:
26
27  bazel run pw_transfer/integration_test:cross_language_small_test -- \
28      SmallTransferIntegrationTest.test_single_byte_client_write_1_java
29
30"""
31
32import itertools
33from parameterized import parameterized
34
35from pigweed.pw_transfer.integration_test import config_pb2
36import test_fixture
37from test_fixture import TransferIntegrationTestHarness
38
39_ALL_LANGUAGES = ("cpp", "java", "python")
40_ALL_VERSIONS = (
41    config_pb2.TransferAction.ProtocolVersion.V1,
42    config_pb2.TransferAction.ProtocolVersion.V2,
43)
44_ALL_LANGUAGES_AND_VERSIONS = tuple(
45    itertools.product(_ALL_LANGUAGES, _ALL_VERSIONS)
46)
47
48
49class SmallTransferIntegrationTest(test_fixture.TransferIntegrationTest):
50    # Each set of transfer tests uses a different client/server port pair to
51    # allow tests to be run in parallel.
52    HARNESS_CONFIG = TransferIntegrationTestHarness.Config(
53        server_port=3302, client_port=3303
54    )
55
56    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
57    def test_empty_client_write(self, client_type, protocol_version):
58        payload = b""
59        config = self.default_config()
60        resource_id = 5
61
62        # Packet drops can cause the resource ID for this to be opened/closed
63        # multiple times due to the zero-size transfer. Use a
64        # permanent_resource_id so the retry process can succeed and the harness
65        # won't flake.
66        self.do_single_write(
67            client_type,
68            config,
69            resource_id,
70            payload,
71            protocol_version,
72            permanent_resource_id=True,
73        )
74
75    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
76    def test_null_byte_client_write(self, client_type, protocol_version):
77        payload = b"\0"
78        config = self.default_config()
79        resource_id = 5
80        self.do_single_write(
81            client_type, config, resource_id, payload, protocol_version
82        )
83
84    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
85    def test_single_byte_client_write(self, client_type, protocol_version):
86        payload = b"?"
87        config = self.default_config()
88        resource_id = 5
89        self.do_single_write(
90            client_type, config, resource_id, payload, protocol_version
91        )
92
93    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
94    def test_small_client_write(self, client_type, protocol_version):
95        payload = b"some data"
96        config = self.default_config()
97        resource_id = 5
98        self.do_single_write(
99            client_type, config, resource_id, payload, protocol_version
100        )
101
102    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
103    def test_empty_client_read(self, client_type, protocol_version):
104        payload = b""
105        config = self.default_config()
106        resource_id = 5
107        self.do_single_read(
108            client_type, config, resource_id, payload, protocol_version
109        )
110
111    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
112    def test_null_byte_client_read(self, client_type, protocol_version):
113        payload = b"\0"
114        config = self.default_config()
115        resource_id = 5
116        self.do_single_read(
117            client_type, config, resource_id, payload, protocol_version
118        )
119
120    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
121    def test_single_byte_client_read(self, client_type, protocol_version):
122        payload = b"?"
123        config = self.default_config()
124        resource_id = 5
125        self.do_single_read(
126            client_type, config, resource_id, payload, protocol_version
127        )
128
129    @parameterized.expand(_ALL_LANGUAGES_AND_VERSIONS)
130    def test_small_client_read(self, client_type, protocol_version):
131        payload = b"some data"
132        config = self.default_config()
133        resource_id = 5
134        self.do_single_read(
135            client_type, config, resource_id, payload, protocol_version
136        )
137
138
139if __name__ == '__main__':
140    test_fixture.run_tests_for(SmallTransferIntegrationTest)
141