#!/usr/bin/env python3 # Copyright (C) 2025 by sysmocom - s.f.m.c. GmbH # Author: Vadim Yanitskiy # # All Rights Reserved # # SPDX-License-Identifier: GPL-3.0-or-later # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import logging import argparse import cmd2 import sys import tabulate import urllib.request import http.client import json # local logger for this module log = logging.getLogger(__name__) class RestIface: ''' REST interface for OsmoS1GW ''' HTTPResponse = http.client.HTTPResponse RESTResponse = dict | list[dict] def __init__(self, host: str, port: int): self.url = f'http://{host}:{port}' def send_req(self, method: str, path: str = '', data: dict = {}) -> HTTPResponse: ''' Send an HTTP request to the given endpoint (path) ''' req = urllib.request.Request(f'{self.url}/{path}', method=method) req.add_header('Accept', 'application/json') if data: req.add_header('Content-Type', 'application/json') req.data = json.dumps(data).encode('utf-8') log.debug(f'HTTP {req.method} {req.full_url}') return urllib.request.urlopen(req) def send_get_req(self, path: str, query: dict = {}) -> HTTPResponse: ''' Send an HTTP GET request to the given endpoint (path) ''' if query: path += '?' + urllib.parse.urlencode(query) return self.send_req('GET', path) def send_post_req(self, path: str, data: dict = {}) -> HTTPResponse: ''' Send an HTTP POST request to the given endpoint (path) ''' return self.send_req('POST', path, data) def send_delete_req(self, path: str, data: dict = {}) -> HTTPResponse: ''' Send an HTTP DELETE request to the given endpoint (path) ''' return self.send_req('DELETE', path, data) def fetch_spec(self) -> RESTResponse: ''' Fetch the OpenAPI specification (JSON) ''' with self.send_get_req('swagger/spec.json') as f: return json.load(f) class OsmoS1GWCli(cmd2.Cmd): DESC = 'Interactive CLI for OsmoS1GW' def __init__(self, argv): super().__init__(allow_cli_args=False, include_py=True) if argv.verbose > 0: logging.root.setLevel(logging.DEBUG) self.debug = True self.intro = cmd2.style('Welcome to %s!' % self.DESC, fg=cmd2.Fg.RED) self.default_category = 'Built-in commands' self.prompt = 'OsmoS1GW# ' self.tablefmt = 'github' # default table format for tabulate self.add_settable(cmd2.Settable('tablefmt', str, 'Table format for tabulate', self, choices=tabulate.tabulate_formats)) self.iface = RestIface(argv.HOST, argv.port) def do_fetch_openapi_spec(self, opts): ''' Fetch the OpenAPI specification (JSON), dump as text ''' spec = self.iface.fetch_spec() self.poutput(json.dumps(spec, indent=4)) ap = argparse.ArgumentParser(prog='osmo-s1gw-cli', description=OsmoS1GWCli.DESC) ap.add_argument('-v', '--verbose', action='count', default=0, help='print debug logging') ap.add_argument('-p', '--port', metavar='PORT', type=int, default=8080, help='OsmoS1GW REST port (default: %(default)s)') ap.add_argument('HOST', type=str, nargs='?', default='localhost', help='OsmoS1GW REST host/address (default: %(default)s)') logging.basicConfig( format='\r[%(levelname)s] %(filename)s:%(lineno)d %(message)s', level=logging.INFO) if __name__ == '__main__': argv = ap.parse_args() app = OsmoS1GWCli(argv) sys.exit(app.cmdloop())