mirror of
https://github.com/lowobservable/oec.git
synced 2026-03-05 02:54:50 +00:00
Improve command line argument parsing
This commit is contained in:
Andrew Kay
@@ -1,10 +1,10 @@
|
||||
import sys
|
||||
import os
|
||||
import signal
|
||||
import codecs
|
||||
import logging
|
||||
import argparse
|
||||
from coax import open_serial_interface, TerminalType
|
||||
|
||||
from .args import parse_args
|
||||
from .interface import InterfaceWrapper
|
||||
from .controller import Controller
|
||||
from .device import get_ids, get_features, get_keyboard_description, UnsupportedDeviceError
|
||||
@@ -39,14 +39,6 @@ def _get_keymap(keyboard_description):
|
||||
|
||||
return KEYMAP_3278_TYPEWRITER
|
||||
|
||||
def _get_character_encoding(encoding):
|
||||
try:
|
||||
codecs.lookup(encoding)
|
||||
except LookupError:
|
||||
raise argparse.ArgumentTypeError(f'invalid encoding: {encoding}')
|
||||
|
||||
return encoding
|
||||
|
||||
def _create_device(args, interface, device_address, poll_response):
|
||||
# Read the terminal identifiers.
|
||||
(terminal_id, extended_id) = get_ids(interface, device_address)
|
||||
@@ -91,38 +83,6 @@ def _create_session(args, device):
|
||||
|
||||
raise ValueError('Unsupported emulator')
|
||||
|
||||
def parse_tn3270_host_args(args, parser):
|
||||
elements = args.host.rsplit(':', 1)
|
||||
|
||||
port = None
|
||||
|
||||
if len(elements) > 1:
|
||||
try:
|
||||
port = int(elements[1])
|
||||
except ValueError:
|
||||
parser.error(f'argument host: invalid port value: {elements[1]}')
|
||||
|
||||
if args.port is not None:
|
||||
if port is None:
|
||||
port = args.port
|
||||
|
||||
logger.info('The port argument is deprecated and will be removed in the future, use host:port instead.')
|
||||
else:
|
||||
logger.warning('The port argument is deprecated and will be removed in the future, port from host:port is being used.')
|
||||
|
||||
if port is None:
|
||||
port = 23
|
||||
|
||||
elements = elements[0].split('@', 1)
|
||||
|
||||
host = elements[-1]
|
||||
device_names = None
|
||||
|
||||
if len(elements) > 1:
|
||||
device_names = elements[0].split(',')
|
||||
|
||||
return (host, port, device_names)
|
||||
|
||||
def _signal_handler(number, frame):
|
||||
global CONTROLLER
|
||||
|
||||
@@ -139,35 +99,7 @@ signal.signal(signal.SIGTERM, _signal_handler)
|
||||
def main():
|
||||
global CONTROLLER
|
||||
|
||||
parser = argparse.ArgumentParser(description='IBM 3270 terminal controller')
|
||||
|
||||
parser.add_argument('serial_port', help='serial port')
|
||||
|
||||
subparsers = parser.add_subparsers(dest='emulator', required=True,
|
||||
description='emulator')
|
||||
|
||||
tn3270_parser = subparsers.add_parser('tn3270', description='TN3270 emulator',
|
||||
help='TN3270 emulator')
|
||||
|
||||
tn3270_parser.add_argument('host', metavar='[lu[,lu...]@]host[:port]',
|
||||
help='host and optional port and LUs')
|
||||
tn3270_parser.add_argument('port', nargs='?', type=int, help=argparse.SUPPRESS)
|
||||
|
||||
tn3270_parser.add_argument('--codepage', metavar='encoding', default='ibm037',
|
||||
dest='character_encoding', type=_get_character_encoding)
|
||||
|
||||
if IS_VT100_AVAILABLE:
|
||||
vt100_parser = subparsers.add_parser('vt100', description='VT100 emulator',
|
||||
help='VT100 emulator')
|
||||
|
||||
vt100_parser.add_argument('command', help='host process')
|
||||
vt100_parser.add_argument('command_args', nargs=argparse.REMAINDER,
|
||||
help='host process arguments')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.emulator == 'tn3270':
|
||||
(args.host, args.port, args.device_names) = parse_tn3270_host_args(args, parser)
|
||||
args = parse_args(sys.argv[1:], IS_VT100_AVAILABLE)
|
||||
|
||||
create_device = lambda interface, device_address, poll_response: _create_device(args, interface, device_address, poll_response)
|
||||
create_session = lambda device: _create_session(args, device)
|
||||
|
||||
96
oec/args.py
Normal file
96
oec/args.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""
|
||||
oec.args
|
||||
~~~~~~~~
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import codecs
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger('oec.args')
|
||||
|
||||
def parse_args(args, is_vt100_available):
|
||||
parser = argparse.ArgumentParser(description='IBM 3270 terminal controller')
|
||||
|
||||
parser.add_argument('serial_port', help='serial port')
|
||||
|
||||
subparsers = parser.add_subparsers(dest='emulator', required=True,
|
||||
description='emulator')
|
||||
|
||||
tn3270_parser = subparsers.add_parser('tn3270', description='TN3270 emulator',
|
||||
help='TN3270 emulator')
|
||||
|
||||
tn3270_parser.add_argument('host', metavar='[lu[,lu...]@]host[:port]',
|
||||
help='host and optional port and LUs')
|
||||
tn3270_parser.add_argument('port', nargs='?', type=int, help=argparse.SUPPRESS)
|
||||
|
||||
tn3270_parser.add_argument('--codepage', metavar='encoding', default='ibm037',
|
||||
dest='character_encoding', type=get_character_encoding)
|
||||
|
||||
if is_vt100_available:
|
||||
vt100_parser = subparsers.add_parser('vt100', description='VT100 emulator',
|
||||
help='VT100 emulator')
|
||||
|
||||
vt100_parser.add_argument('command', help='host process')
|
||||
vt100_parser.add_argument('command_args', nargs=argparse.REMAINDER,
|
||||
help='host process arguments')
|
||||
|
||||
args = parser.parse_args(args)
|
||||
|
||||
if args.emulator == 'tn3270':
|
||||
(args.host, args.port, args.device_names) = parse_tn3270_host_args(args, parser)
|
||||
|
||||
return args
|
||||
|
||||
def get_character_encoding(encoding):
|
||||
try:
|
||||
codecs.lookup(encoding)
|
||||
except LookupError:
|
||||
raise argparse.ArgumentTypeError(f'invalid encoding: {encoding}')
|
||||
|
||||
return encoding
|
||||
|
||||
def parse_tn3270_host_args(args, parser):
|
||||
elements = args.host.rsplit(':', 1)
|
||||
|
||||
port = None
|
||||
|
||||
if len(elements) > 1:
|
||||
try:
|
||||
port = int(elements[1])
|
||||
except ValueError:
|
||||
parser.error(f'argument host: invalid port: {elements[1]}')
|
||||
|
||||
if not is_valid_port(port):
|
||||
parser.error(f'argument host: invalid port: {port}')
|
||||
|
||||
if args.port is not None:
|
||||
if port is None:
|
||||
if not is_valid_port(args.port):
|
||||
parser.error(f'argument port: invalid port: {args.port}')
|
||||
|
||||
port = args.port
|
||||
|
||||
logger.info('The port argument is deprecated and will be removed in the future, use host:port instead.')
|
||||
else:
|
||||
logger.warning('The port argument is deprecated and will be removed in the future, port from host:port is being used.')
|
||||
|
||||
if port is None:
|
||||
port = 23
|
||||
|
||||
elements = elements[0].split('@', 1)
|
||||
|
||||
host = elements[-1].strip()
|
||||
|
||||
if not host:
|
||||
parser.error(f'argument host: host name is required: {args.host}')
|
||||
|
||||
device_names = None
|
||||
|
||||
if len(elements) > 1:
|
||||
device_names = elements[0].split(',')
|
||||
|
||||
return (host, port, device_names)
|
||||
|
||||
def is_valid_port(port):
|
||||
return port >= 1 and port <= 65535
|
||||
87
tests/test_args.py
Normal file
87
tests/test_args.py
Normal file
@@ -0,0 +1,87 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import context
|
||||
|
||||
from oec.args import parse_args
|
||||
|
||||
class ParseArgsTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
patcher = patch('argparse.ArgumentParser.error')
|
||||
|
||||
self.parser_error = patcher.start()
|
||||
|
||||
self.addCleanup(patch.stopall)
|
||||
|
||||
def test_tn3270_host_only(self):
|
||||
args = parse_args(['/dev/ttyACM0', 'tn3270', 'host'], False)
|
||||
|
||||
self.assertEqual(args.emulator, 'tn3270')
|
||||
self.assertEqual(args.host, 'host')
|
||||
self.assertEqual(args.port, 23)
|
||||
self.assertIsNone(args.device_names)
|
||||
|
||||
def test_tn3270_host_and_port(self):
|
||||
args = parse_args(['/dev/ttyACM0', 'tn3270', 'host:3270'], False)
|
||||
|
||||
self.assertEqual(args.emulator, 'tn3270')
|
||||
self.assertEqual(args.host, 'host')
|
||||
self.assertEqual(args.port, 3270)
|
||||
self.assertIsNone(args.device_names)
|
||||
|
||||
def test_tn3270_host_and_device_names(self):
|
||||
args = parse_args(['/dev/ttyACM0', 'tn3270', 'lu1@host'], False)
|
||||
|
||||
self.assertEqual(args.emulator, 'tn3270')
|
||||
self.assertEqual(args.host, 'host')
|
||||
self.assertEqual(args.port, 23)
|
||||
self.assertEqual(args.device_names, ['lu1'])
|
||||
|
||||
def test_tn3270_missing_host(self):
|
||||
for arg in [':3270', 'lu1@', 'lu1@:3270']:
|
||||
with self.subTest(arg=arg):
|
||||
self.parser_error.reset_mock()
|
||||
|
||||
parse_args(['/dev/ttyACM0', 'tn3270', arg], False)
|
||||
|
||||
self.parser_error.assert_called_once()
|
||||
|
||||
self.assertEqual(self.parser_error.call_args.args[0], f'argument host: host name is required: {arg}')
|
||||
|
||||
def test_tn3270_invalid_port(self):
|
||||
for port in ['-1', '0', '100000']:
|
||||
with self.subTest(port=port):
|
||||
self.parser_error.reset_mock()
|
||||
|
||||
parse_args(['/dev/ttyACM0', 'tn3270', 'host:' + port], False)
|
||||
|
||||
self.parser_error.assert_called_once()
|
||||
|
||||
self.assertEqual(self.parser_error.call_args.args[0], f'argument host: invalid port: {port}')
|
||||
|
||||
def test_tn3270_deprecated_port(self):
|
||||
args = parse_args(['/dev/ttyACM0', 'tn3270', 'host', '3270'], False)
|
||||
|
||||
self.assertEqual(args.emulator, 'tn3270')
|
||||
self.assertEqual(args.host, 'host')
|
||||
self.assertEqual(args.port, 3270)
|
||||
self.assertIsNone(args.device_names)
|
||||
|
||||
def test_tn3270_deprecated_port_overridden_by_new_port(self):
|
||||
args = parse_args(['/dev/ttyACM0', 'tn3270', 'host:3270', '9999'], False)
|
||||
|
||||
self.assertEqual(args.emulator, 'tn3270')
|
||||
self.assertEqual(args.host, 'host')
|
||||
self.assertEqual(args.port, 3270)
|
||||
self.assertIsNone(args.device_names)
|
||||
|
||||
def test_tn3270_invalid_deprecated_port(self):
|
||||
for port in ['-1', '0', '100000']:
|
||||
with self.subTest(port=port):
|
||||
self.parser_error.reset_mock()
|
||||
|
||||
parse_args(['/dev/ttyACM0', 'tn3270', 'host', port], False)
|
||||
|
||||
self.parser_error.assert_called_once()
|
||||
|
||||
self.assertEqual(self.parser_error.call_args.args[0], f'argument port: invalid port: {port}')
|
||||
Reference in New Issue
Block a user