/openbmc/linux/tools/testing/selftests/drivers/net/mlxsw/ |
H A D | sharedbuffer_configuration.py | 2 # SPDX-License-Identifier: GPL-2.0 19 - random size 22 - random pool number 23 - random threshold 26 - random threshold 28 def __init__(self, pools): 33 def _cell_size(self): 36 def _get_static_size(self, th): 37 # For threshold of 16, this works out to be about 12MB on Spectrum-1, 38 # and about 17MB on Spectrum-2. [all …]
|
/openbmc/linux/tools/testing/selftests/net/ |
H A D | devlink_port_split.py | 2 # SPDX-License-Identifier: GPL-2.0 12 # Test port split configuration using devlink-port lanes attribute. 21 # Kselftest framework requirement - SKIP code is 4 23 Port = collections.namedtuple('Port', 'bus_info name') variable 26 def run_command(cmd, should_fail=False): 49 def get_if_names(dev): 52 Return: Array of tuples (bus_info/port, if_name). 57 cmd = "devlink -j port show" 60 ports = json.loads(stdout)['port'] 64 for port in ports: [all …]
|
/openbmc/openbmc/poky/bitbake/lib/prserv/ |
H A D | serv.py | 4 # SPDX-License-Identifier: GPL-2.0-only 24 def __init__(self, socket, server): 29 "get-pr": self.handle_get_pr, 30 "test-pr": self.handle_test_pr, 31 "test-package": self.handle_test_package, 32 "max-package-pr": self.handle_max_package_pr, 33 "import-one": self.handle_import_one, 35 "is-readonly": self.handle_is_readonly, 38 def validate_proto_version(self): 41 async def dispatch_message(self, msg): [all …]
|
/openbmc/openbmc/poky/bitbake/lib/bb/server/ |
H A D | xmlrpcclient.py | 4 # Copyright (C) 2006 - 2007 Michael 'Mickey' Lauer 5 # Copyright (C) 2006 - 2008 Richard Purdie 7 # SPDX-License-Identifier: GPL-2.0-only 18 def __init__(self, timeout): 24 def make_connection(self, host): 26 #HTTP/1.1 keep-alive. 36 def set_connection_token(self, token): 39 def send_content(self, h, body): 41 h.putheader("Bitbake-token", self.connection_token) 44 def _create_server(host, port, timeout = 60): argument [all …]
|
H A D | xmlrpcserver.py | 4 # Copyright (C) 2006 - 2007 Michael 'Mickey' Lauer 5 # Copyright (C) 2006 - 2008 Richard Purdie 7 # SPDX-License-Identifier: GPL-2.0-only 20 # This request handler checks if the request has a "Bitbake-token" header 22 # "Bitbake-token" field (this comes from the server). If the two are not 27 def __init__(self, request, client_address, server): 31 def do_POST(self): 33 remote_token = self.headers["Bitbake-token"] 45 def report_503(self): 48 self.send_header("Content-type", "text/plain") [all …]
|
/openbmc/openbmc/poky/bitbake/lib/bb/asyncrpc/ |
H A D | serv.py | 4 # SPDX-License-Identifier: GPL-2.0-only 21 def process(self, msg, kwargs): 30 def __init__(self, socket, proto_name, logger): 44 async def close(self): 47 async def handle_headers(self, headers): 50 async def process_requests(self): 81 if self.client_headers.get("needs-headers", "false") == "true": 95 {"invoke-error": {"message": str(e)}} 109 async def dispatch_message(self, msg): 117 async def handle_ping(self, request): [all …]
|
H A D | client.py | 4 # SPDX-License-Identifier: GPL-2.0-only 33 def parse_address(addr): 39 m = re.match(r"\[(?P<host>[^\]]*)\]:(?P<port>\d+)$", addr) 42 port = m.group("port") 44 host, port = addr.split(":") 46 return (ADDR_TYPE_TCP, (host, int(port))) 50 def __init__( 69 async def connect_tcp(self, address, port): argument 70 async def connect_sock(): 71 reader, writer = await asyncio.open_connection(address, port) [all …]
|
/openbmc/qemu/tests/avocado/ |
H A D | vnc.py | 9 # later. See the COPYING file in the top-level directory. 22 def check_bind(port: int) -> bool: 25 sock.bind((VNC_ADDR, port)) 32 def check_connect(port: int) -> bool: 35 sock.connect((VNC_ADDR, port)) 42 def find_free_ports(count: int) -> List[int]: 44 for port in range(VNC_PORT_START, VNC_PORT_END): 45 if check_bind(port): 46 result.append(port) 58 def test_no_vnc(self): [all …]
|
H A D | migration.py | 10 # later. See the COPYING file in the top-level directory. 32 def migration_finished(vm): 33 return vm.cmd('query-migrate')['status'] in ('completed', 'failed') 35 def assert_migration(self, src_vm, dst_vm): 44 self.assertEqual(src_vm.cmd('query-migrate')['status'], 'completed') 45 self.assertEqual(dst_vm.cmd('query-migrate')['status'], 'completed') 46 self.assertEqual(dst_vm.cmd('query-status')['status'], 'running') 47 self.assertEqual(src_vm.cmd('query-status')['status'],'postmigrate') 49 def do_migrate(self, dest_uri, src_uri=None): 50 dest_vm = self.get_vm('-incoming', dest_uri) [all …]
|
H A D | reverse_debugging.py | 9 # later. See the COPYING file in the top-level directory. 38 def run_vm(self, record, shift, args, replay_path, image_path, port): argument 48 vm.add_args('-gdb', 'tcp::%d' % port, '-S') 49 vm.add_args('-icount', 'shift=%s,rr=%s,rrfile=%s,rrsnapshot=init' % 51 '-net', 'none') 52 vm.add_args('-drive', 'file=%s,if=none' % image_path) 63 def get_reg_le(g, reg): 66 for i in range(len(res))[-2::-2]: 71 def get_reg_be(g, reg): 75 def get_reg(self, g, reg): [all …]
|
/openbmc/openbmc/poky/meta/lib/oeqa/utils/ |
H A D | httpserver.py | 4 # SPDX-License-Identifier: MIT 16 def server_start(self, root_dir, logger): 23 def log_message(self, format_str, *args): 28 def __init__(self, root_dir, host='', port=0, logger=None): argument 31 self.port = port 37 def start(self): 42 self.server = HTTPServer((self.host, self.port), HTTPRequestHandler) 43 if self.port == 0: 44 self.port = self.server.server_port 47 def handle_error(self, request, client_address): [all …]
|
/openbmc/openbmc/poky/bitbake/lib/bb/tests/support/ |
H A D | httpserver.py | 2 # SPDX-License-Identifier: MIT 15 def server_start(self, root_dir, logger): 21 def log_message(self, format_str, *args): 26 def __init__(self, root_dir, host='', port=0, logger=None): argument 29 self.port = port 34 def start(self): 40 self.server = HTTPServer((self.host, self.port), HTTPRequestHandler) 41 if self.port == 0: 42 self.port = self.server.server_port 54 self.logger.info("Started HTTPService on %s:%s" % (self.host, self.port)) [all …]
|
/openbmc/qemu/tests/avocado/avocado_qemu/ |
H A D | __init__.py | 9 # later. See the COPYING file in the top-level directory. 35 def has_cmd(name, args=None): 39 @skipUnless(*has_cmd('sudo -n', ('sudo', '-n', 'true'))) 40 def test_something_that_needs_sudo(self): 50 exitcode = -1 60 def has_cmds(*cmds): 65 @skipUnless(*has_cmds(('cmd1', ('cmd1', '--some-parameter')), 67 def test_something_that_needs_cmd1_and_cmd2(self): 81 def run_cmd(args): 91 def is_readable_executable_file(path): [all …]
|
/openbmc/openbmc/poky/meta/lib/oeqa/selftest/cases/ |
H A D | prservice.py | 4 # SPDX-License-Identifier: MIT 22 def setUpClass(cls): 29 def cleanup(self): 31 bitbake("-m") 36 def get_pr_version(self, package_name): 39 find_pr = re.search(r"PKGR: r[0-9]+\.([0-9]+)", package_data) 40 …self.assertTrue(find_pr, "No PKG revision found via regex 'PKGR: r[0-9]+\.([0-9]+)' in %s" % packa… 43 def get_task_stamp(self, package_name, recipe_task): 45 prefix = stampdata[-1] 46 package_stamps_path = "/".join(stampdata[:-1]) [all …]
|
/openbmc/openbmc/poky/bitbake/lib/bb/ui/ |
H A D | uievent.py | 2 # Copyright (C) 2006 - 2007 Michael 'Mickey' Lauer 3 # Copyright (C) 2006 - 2007 Richard Purdie 5 # SPDX-License-Identifier: GPL-2.0-only 22 def __init__(self, BBServer, clientinfo=("localhost, 0")): 32 self.host, self.port = server.socket.getsockname() 45 ret = self.BBServer.registerEventHandler(self.host, self.port) 57 "port %d" % (error, self.host, self.port) 72 def getEvent(self): 81 def waitEvent(self, delay): 85 def queue_event(self, event): [all …]
|
/openbmc/openbmc/meta-arm/meta-arm/lib/fvp/ |
H A D | runner.py | 11 def cli_from_config(config, terminal_choice): 13 if config["fvp-bindir"]: 14 cli.append(os.path.join(config["fvp-bindir"], config["exe"])) 19 cli.extend(["--parameter", f"{param}={value}"]) 22 cli.extend(["--data", value]) 25 cli.extend(["--application", f"{param}={value}"]) 31 # cli.extend(["--parameter", f"{terminal}.mode=raw"]) 32 …cli.extend(["--parameter", f"{terminal}.terminal_command={terminals[terminal_choice].command.forma… 35 cli.extend(["--parameter", f"{terminal}.start_telnet=0"]) 41 def check_telnet(): [all …]
|
/openbmc/qemu/python/qemu/qmp/ |
H A D | legacy.py | 14 # Copyright (C) 2009-2022 Red Hat Inc. 21 # the COPYING file in the top-level directory. 54 # 'command-returns-exceptions', it can actually be anything. 62 Unable to parse socket address: Port was non-numerical. 72 in the form ( address, port ) for a TCP connection, or an 79 def __init__(self, 99 def _sync( 101 ) -> _T: 106 def _get_greeting(self) -> Optional[QMPMessage]: 108 # pylint: disable=protected-access [all …]
|
/openbmc/qemu/tests/qemu-iotests/ |
H A D | nbd-fault-injector.py | 2 # NBD server - fault injection utility 5 # [inject-error "disconnect-neg1"] 11 # name, so give each [inject-error] a unique name. 13 # inject-error options: 14 # event - name of the trigger event 15 # "neg1" - first part of negotiation struct 16 # "export" - export struct 17 # "neg2" - second part of negotiation struct 18 # "request" - NBD request struct 19 # "reply" - NBD reply struct [all …]
|
/openbmc/pldm/tools/visualize-pdr/ |
H A D | pldm_visualise_pdrs.py | 24 def __init__(self, stdout, stderr): 37 def wait(self): 47 def exec_command(self, cmd): 50 def close(self): 58 def __init__(self, stdout, stderr): 61 def wait(self): 69 def __init__(self, hostname, uname, passwd, port, **kw): argument 77 port: ssh port of BMC 84 hostname, username=uname, password=passwd, port=port, **kw 87 def exec_command(self, cmd): [all …]
|
/openbmc/openbmc/meta-openembedded/meta-oe/recipes-support/libteam/libteam/ |
H A D | 0001-team_basic_test.py-switch-to-python3.patch | 8 Upstream-Status: Submitted [https://github.com/jpirko/libteam/pull/63] 10 Signed-off-by: Mingli Yu <mingli.yu@windriver.com> 11 --- 12 scripts/team_basic_test.py | 28 ++++++++++++++-------------- 13 1 file changed, 14 insertions(+), 14 deletions(-) 15 diff --git a/scripts/team_basic_test.py b/scripts/team_basic_test.py 17 --- a/scripts/team_basic_test.py 19 @@ -1,4 +1,4 @@ 20 -#! /usr/bin/env python 25 @@ -32,11 +32,11 @@ def usage(): [all …]
|
/openbmc/openbmc/meta-arm/meta-arm/lib/oeqa/controllers/ |
H A D | fvp.py | 19 For compatibility with OE-core test cases, this target's start() method 25 def __init__(self, logger, target_ip, server_ip, timeout=300, user='root', 26 port=None, dir_image=None, rootfs=None, bootlog=None, **kwargs): argument 27 super().__init__(logger, target_ip, server_ip, timeout, user, port) 41 def transition(self, state, timeout=10*60): 66 …(b"\n".join(self.before(OEFVPTarget.DEFAULT_CONSOLE).splitlines()[-200:]).decode("utf-8", errors="… 72 def start(self, **kwargs): 73 # No-op - put the FVP in the required state lazily 76 def stop(self, **kwargs): 79 def run(self, cmd, timeout=None): [all …]
|
/openbmc/openbmc-test-automation/lib/external_intf/ |
H A D | management_console_utils.py | 12 bmc_rec_prop = ["hostname", "address", "port", "txt"] 16 def __init__(self, exc_value): 19 def __str__(self): 23 def Check_bmc_record_exists(bmc_records, bmc_ip): 43 def validate_bmc_properties( 51 bmc_prop BMC property (e.g. hostname, address, port). 69 def bmc_record_validation(bmc_rec_valid): 97 def bmc_inventory(service_type, bmc_inv_record): 113 [port]: XXX 129 line.split("=")[-1].strip() [all …]
|
/openbmc/qemu/tests/functional/acpi-bits/bits-tests/ |
H A D | smilatency.py2 | 4 # SPDX-License-Identifier: BSD-3-Clause 39 def register_tests(): 44 def smi_latency(): 78 deltas = (bits.format_tsc(t2 - t1) for t1,t2 in zip(bin.times, bin.times[1:])) 86 def test_with_usb_disabled(): 90 def average_io_smi(port, value, count): 91 def f(): 93 bits.outb(port, value) 94 return bits.rdtsc() - tsc_start 98 def time_io_smi(port=0xb2, value=0, count=1000): [all …]
|
/openbmc/openbmc/poky/bitbake/bin/ |
H A D | bitbake-hashclient | 5 # SPDX-License-Identifier: GPL-2.0-only 28 def __init__(self, *args, **kwargs): 31 def __enter__(self): 34 def __exit__(self, *args, **kwargs): 37 def update(self): 48 def print_user(u): 56 def main(): 57 def handle_get(args, client): 65 def handle_get_outhash(args, client): 73 def handle_stats(args, client): [all …]
|
/openbmc/openbmc/poky/meta/lib/oeqa/runtime/cases/ |
H A D | apt.py | 4 # SPDX-License-Identifier: MIT 15 def pkg(self, command, expected = 0): 16 command = 'apt-get %s' % command 25 def setUpClass(cls): 28 '0.0.0.0', port=cls.tc.target.server_port, 33 def tearDownClass(cls): 36 def setup_source_config_for_package_install(self): 37 apt_get_source_server = 'http://%s:%s/' % (self.tc.target.server_ip, self.repo_server.port) 39 …self.target.run('cd %s; echo deb [ allow-insecure=yes ] %s/all ./ > sources.list' % (apt_get_sourc… 41 def setup_source_config_for_package_install_signed(self): [all …]
|