Commit 25b1eed8 authored by Yann Garcia's avatar Yann Garcia
Browse files

Enahnce Upper Tester

parent 61a1d038
Loading
Loading
Loading
Loading
+12 −1
Changes for ttcn/AtsNGAP/NGAP_UtDispatcher.ttcn: 12 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -56,12 +56,15 @@ module NGAP_UtDispatcher {
     * @param  p_setVerdict When true, set the verdict: pass if the UT confirms success, inconc if
     *         it reports a failure, does not confirm in time or the request cannot be mapped.
     *         Requests without confirmation (NO_CNF_REQUIRED) leave the verdict unchanged.
     *         A failure is logged with the confirmation's ResultString, which holds the AT
     *         transcript of the request; it is also the inconc verdict reason.
     */
    function f_UT_HandleRequest(in UT_COMMON_CMD p_cmd, in boolean p_setVerdict := false) runs on NGNASComponent {
        var Ut_System_Cmd_Type v_sys;
        var UT_COMMON_CNF v_cnf;
        timer t_cnf := PX_UT_CNF_TIMEOUT;
        var boolean v_cnfExpected := f_UT_CommandRequiresCnf(p_cmd.CnfReq);
        var charstring v_reason := "no result string";

        if (not f_UT_MapCommand(p_cmd, v_sys)) {
            log("f_UT_HandleRequest: missing parameter for ", p_cmd.Cmd);
@@ -80,9 +83,17 @@ module NGAP_UtDispatcher {
        alt {
            [] utPort.receive(car_UT_COMMON_CNF) -> value v_cnf {
                t_cnf.stop;
                if (not v_cnf.Result) {
                    // On failure, ResultString carries the AT transcript of the request
                    // ("AT> <command> -> <response>" lines), then the reason ("+UT: ..." lines)
                    if (ispresent(v_cnf.ResultString)) {
                        v_reason := v_cnf.ResultString;
                    }
                    log("f_UT_HandleRequest: ", p_cmd.Cmd, " failed: ", v_reason);
                }
                if (p_setVerdict) {
                    if (v_cnf.Result) { setverdict(pass, "UT command succeeded"); }
                    else { setverdict(inconc, "UT command failed"); }
                    else { setverdict(inconc, "UT command failed: " & v_reason); }
                }
                utMtcPtcPort.send(v_cnf);
            }
+38 −1
Changes for upper_tester/README.md: 38 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -26,6 +26,43 @@ Reply (one line):
  `+UT: modem restarted`. This takes the time of a modem boot.
- If the modem disappears while a command is in flight, that command fails at once with
  `+UT: modem disconnected`; later commands wait for the device to come back.
- When a request fails after running AT commands, `result_string` is its AT transcript, one
  command per line, then the `+UT:` lines giving the reason. Identical commands in a row (e.g.
  attach polling) are shown once with a count, and the oldest lines are dropped beyond 3000
  characters:
  ```
  AT> AT+CGATT? -> +CGATT: 1 | OK
  AT> AT+CGACT? -> +CGACT: 1,0 | OK
  AT> AT+CGACT=1,1 -> +CME ERROR: 30
  +UT: data call rejected: cid=1
  ```
  A successful request keeps the plain response text, which the test system may parse.

## Modem profiles

Everything that depends on the modem model or firmware is in a profile (`profiles.py`): AT
timeouts, the commands run after (re)start, restart detection and timings, commands refused
before they are sent, the default of the data call `NETDEV` parameter, and MMI commands mapped
differently on this modem. Select one with `--profile` (or `profile = ...` in the config file):

| Profile          | Modem                                              |
|------------------|----------------------------------------------------|
| `generic`        | Any 3GPP TS 27.007 modem, standard commands only   |
| `rm520n-gl`      | Quectel RM520N-GL (default): no `AT+QNETDEVCTL`; `AT+COPS=2` is refused, because it leaves the modem in `SIM failure` |
| `quectel-netdev` | Quectel modems with `AT+QNETDEVCTL` (RG500U, RG801H, EC200T): `NETDEV=1` by default |

`--list-profiles` prints them. For another modem, write a TOML file and pass its path:
```toml
base = "rm520n-gl"          # profile to start from (default "generic")
name = "rm520n-gl-lab"
restart_back_timeout = 120.0
[timeouts]                  # command prefix = seconds, checked first
"AT+COPS" = 300
[refused]                   # regular expression = reason
'AT\+QPOWD.*' = "powers the modem off"
[mmi]                       # MMI command = AT commands
DEREGISTER = ["AT+CFUN=4"]
```

## Data call (`ACTIVATE_BEARER` / `DEACTIVATE_BEARER`)

@@ -56,7 +93,7 @@ On failure it ends with a `+UT:` line giving the reason.
The call is dialled with `AT+CGACT=1,<cid>`: the network sees the session, but the host has no
data path through the modem's network interface. `AT+QNETDEVCTL`, which `quectel-CM` uses to
also connect that interface, is not supported by the RM520N-GL and its firmware; `NETDEV=1`
asks for it on a modem that has it. The host network interface is never configured here
asks for it on a modem that has it (the default with the `quectel-netdev` profile). The host network interface is never configured here
(`quectel-CM` runs `udhcpc` for that, as root).

## Run / test
+21 −5
Changes for upper_tester/__main__.py: 21 added lines, 5 removed lines.
Original line number Diff line number Diff line
@@ -29,6 +29,7 @@ import asyncio
import logging
import tomllib

from . import profiles
from .engine import Engine
from .modem import AtModem
from .server import UtServer
@@ -38,16 +39,19 @@ def main() -> None:
    """!
    @brief  Parse arguments/config, open the modem and run the server until Ctrl-C.
    @details Options: `--config` (a TOML file whose keys are option names: `device`, `baud`,
             `host`, `port`, `allow`, `log_level`; overrides the corresponding
             `host`, `port`, `allow`, `log_level`, `profile`; overrides the corresponding
             command-line default for any key it sets), `--device` (default
             `/dev/ttyUSB2`), `--baud` (default 115200), `--host` (default `0.0.0.0`),
             `--port` (default 12100), `--allow` (repeatable, an allowed client CIDR
             network; see server.UtServer), `--log-level` (default `INFO`),
             `--open-retries`/`--open-retry-delay` (default 20 attempts, 3 s apart:
             long enough for the modem to come back after a restart).
             long enough for the modem to come back after a restart), `--profile` (the
             modem profile: a built-in name or a `.toml` file, see profiles.py; default
             profiles.DEFAULT_PROFILE), `--list-profiles` (print the built-in profiles
             and exit).
    """
    ap = argparse.ArgumentParser(prog="upper_tester")
    ap.add_argument("--config", help="TOML file; keys = option names (device, baud, host, port, allow, log_level)")
    ap.add_argument("--config", help="TOML file; keys = option names (device, baud, host, port, allow, log_level, profile)")
    ap.add_argument("--device", default="/dev/ttyUSB2")
    ap.add_argument("--baud", type=int, default=115200)
    ap.add_argument("--host", default="0.0.0.0")
@@ -56,7 +60,14 @@ def main() -> None:
    ap.add_argument("--log-level", default="INFO")
    ap.add_argument("--open-retries", type=int, default=20, help="attempts to open the device before giving up")
    ap.add_argument("--open-retry-delay", type=float, default=3.0, help="seconds between open attempts")
    ap.add_argument("--profile", default=profiles.DEFAULT_PROFILE,
                    help=f"modem profile: built-in name or .toml file (default {profiles.DEFAULT_PROFILE})")
    ap.add_argument("--list-profiles", action="store_true", help="print the built-in modem profiles and exit")
    args = ap.parse_args()
    if args.list_profiles:
        for name, prof in profiles.BUILTIN.items():
            print(f"{name:16} {prof.description}")
        return
    if args.config:
        with open(args.config, "rb") as f:
            for k, v in tomllib.load(f).items():
@@ -64,9 +75,14 @@ def main() -> None:
    logging.basicConfig(level=args.log_level.upper(), format="%(asctime)s %(levelname)s %(name)s: %(message)s")

    log = logging.getLogger("ut")
    modem = AtModem(args.device, args.baud, args.open_retries, args.open_retry_delay)
    try:
        profile = profiles.load(args.profile)
    except (OSError, ValueError) as e:
        ap.error(f"--profile: {e}")
    log.info("modem profile: %s (%s)", profile.name, profile.description)
    modem = AtModem(args.device, args.baud, args.open_retries, args.open_retry_delay, profile)
    modem.open()
    modem.init()  # no echo, verbose +CME ERROR causes; see modem.INIT_COMMANDS
    modem.init()  # no echo, verbose +CME ERROR causes; see profiles.ModemProfile.init_commands

    async def run() -> None:
        """! @brief Start the TCP server and serve until cancelled. """
+10 −7
Changes for upper_tester/datacall.py: 10 added lines, 7 removed lines.
Original line number Diff line number Diff line
@@ -17,7 +17,8 @@

          `AT+QNETDEVCTL`, which quectel-CM uses to also connect the modem's network interface
          to the session, is not supported by the RM520N-GL and its firmware (it answers
          `ERROR`), so it is only used on request, with `NETDEV=1`, for a modem that has it:
          `ERROR`), so it is only used with `NETDEV=1`, given in the request or as the default of
          the modem profile (profiles.ModemProfile.netdev), for a modem that has it:
          - `AT+QNETDEVSTATUS=?` answers OK: `AT+QNETDEVCTL=<cid>,1,1`, then poll
            `AT+QNETDEVSTATUS=<cid>` until it reports an address;
          - otherwise: `AT+CGACT=1,<cid>` (unless already active), then
@@ -37,7 +38,7 @@
          | `AUTH`           | 0 none, 1 PAP, 2 CHAP, 3 MsChapV2                    | 0, or 2 with a password |
          | `IPTYPE`         | `IPV4`, `IPV6` or `IPV4V6`                           | `IPV4`  |
          | `ATTACH_TIMEOUT` | Seconds to wait for attach; 0 skips the check        | 30      |
          | `NETDEV`         | 1 to dial with `AT+QNETDEVCTL`, 0 with `AT+CGACT` only | 0     |
          | `NETDEV`         | 1 to dial with `AT+QNETDEVCTL`, 0 with `AT+CGACT` only | modem profile (profiles.ModemProfile.netdev) |
"""
import logging
import time
@@ -136,15 +137,17 @@ class Profile:
        self.netdev = netdev

    @classmethod
    def from_params(cls, params: dict[str, str]) -> "Profile":
    def from_params(cls, params: dict[str, str], netdev: bool = False) -> "Profile":
        """!
        @brief  Build a profile from the request parameters.
        @param  params  `Name -> Value`, as received; names are matched case-insensitively.
        @param  netdev  Default of `NETDEV` when the request does not give it.
        @return The validated profile.
        @throws BadParameter if a value is out of range, or holds a character that cannot be
                sent inside a quoted AT string (double quote or control character).
        """
        p = {k.upper(): v for k, v in params.items()}
        p.setdefault("NETDEV", "1" if netdev else "0")
        try:
            cid = int(p.get("CID", "1"))
            attach_timeout = float(p.get("ATTACH_TIMEOUT", DEFAULT_ATTACH_TIMEOUT))
@@ -154,7 +157,7 @@ class Profile:
            raise BadParameter(f"CID {cid} not in 1..15")
        if attach_timeout < 0:
            raise BadParameter(f"ATTACH_TIMEOUT {attach_timeout} is negative")
        if p.get("NETDEV", "0") not in ("0", "1"):
        if p["NETDEV"] not in ("0", "1"):
            raise BadParameter(f"NETDEV {p['NETDEV']} not 0 or 1")
        texts = {name: p.get(name, "") for name in ("APN", "USER", "PASSWORD")}
        for name, text in texts.items():
@@ -168,7 +171,7 @@ class Profile:
        if auth is None:
            raise BadParameter(f"AUTH {p['AUTH']} not in {sorted(AUTHS)}")
        return cls(cid, p.get("APN"), texts["USER"], texts["PASSWORD"], auth, iptype, attach_timeout,
                   p.get("NETDEV", "0") == "1")
                   p["NETDEV"] == "1")


class _At:
@@ -298,7 +301,7 @@ def activate(modem, params: dict[str, str], timeout: float | None = None, on_wri
    """
    at = _At(modem, timeout, on_written)
    try:
        prof = Profile.from_params(params)
        prof = Profile.from_params(params, modem.profile.netdev)
    except BadParameter as e:
        at.note(f"bad parameter: {e}")
        return at.result(False)
@@ -364,7 +367,7 @@ def deactivate(modem, params: dict[str, str], timeout: float | None = None, on_w
    """
    at = _At(modem, timeout, on_written)
    try:
        prof = Profile.from_params(params)
        prof = Profile.from_params(params, modem.profile.netdev)
    except BadParameter as e:
        at.note(f"bad parameter: {e}")
        return at.result(False)
+1 −0
Changes for upper_tester/deploy/upper-tester.toml: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -4,3 +4,4 @@ host = "0.0.0.0"
port = 12100
allow = ["192.168.1.0/24"]   # test-system host(s)
log_level = "INFO"
profile = "rm520n-gl"     # modem profile: built-in name or .toml file (see profiles.py)
Loading