Enahnce UpperTester to support data call (50caec0e) · Commits · INT - Core Network and Interoperability Testing / 5G Core / NAS

upper_tester/README.md

+30 −0

Changes for upper_tester/README.md: 30 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,6 +22,36 @@ Reply (one line):
		- MMI commands are mapped to AT sequences in `mmi.py` (`MMI_TABLE`); unknown ones reply `result=false`.
		- Unsolicited result codes are only logged.

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

		These two MMI commands set up and release a data call (PDP context / PDU session) with AT
		commands, following the AT backend of Quectel's `quectel-CM`. Unlike that tool, a failed
		call is reported once and never redialled. See `datacall.py`.

		```json
		{"id":20,"type":"MMI","cmd":"ACTIVATE_BEARER","params":[{"name":"APN","value":"internet"}],"cnf":"CNF_REQUIRED"}
		{"id":21,"type":"MMI","cmd":"DEACTIVATE_BEARER","params":[{"name":"CID","value":"1"}],"cnf":"CNF_REQUIRED"}
		```

		\| Parameter \| Meaning \| Default \|
		\|------------------\|------------------------------------------------------\|---------\|
		\| `CID` \| PDP context identifier, 1..15 \| 1 \|
		\| `APN` \| APN/DNN; when absent the profile is left untouched \| - \|
		\| `USER` \| User name \| empty \|
		\| `PASSWORD` \| Password \| empty \|
		\| `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 \|

		Parameter names are case-insensitive. `ACTIVATE_BEARER` replies `result=true` only once the
		modem reports an IP address; the reply then ends with `+UT: data call connected: cid=1 ip=...`.
		On failure it ends with a `+UT:` line giving the reason.

		The call is dialled with `AT+QNETDEVCTL`, which needs the modem in ECM/RNDIS/NCM mode
		(`AT+QCFG="usbnet"`). If the modem refuses it, only the context is activated (`AT+CGACT`):
		the network sees the session but the host has no data path, and the reply says so. The host
		network interface is never configured here (`quectel-CM` runs `udhcpc` for that, as root).

		## Run / test
		From the nas project root, run the tests to validate the installation:
		```sh

upper_tester/datacall.py

0 → 100644

+364 −0

Changes for upper_tester/datacall.py: 364 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""!
		@file
		@brief Data call (PDP context / PDU session) setup and release over AT commands.
		@details Follows the AT backend of Quectel's QConnectManager (`quectel-CM`, atc.c), but as
		one explicit, test-controlled operation: no background monitoring and no automatic
		redial, so a rejected data call is reported once and never retried on its own.

		@ref activate runs, in order:
		-# Profile: when an `APN` is given, read the context with `AT+QICSGP=<cid>` and
		rewrite it only if it differs (`AT+QICSGP`, falling back to `AT+CGDCONT`).
		-# Attach: wait until the modem is packet-service attached (`AT+CGATT?`).
		-# Dial, in one of three ways depending on what the modem supports:
		- `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
		`AT+QNETDEVCTL=1,<cid>,1`;
		- `AT+QNETDEVCTL` itself is refused (modem not in ECM/RNDIS/NCM mode): the
		context is activated with `AT+CGACT=1,<cid>` alone. The network then sees the
		session, but the host gets no data path. This fallback is not in quectel-CM,
		which uses QMI or MBIM in that case.
		-# Address: the call counts as connected only once the modem reports an IP
		address (`AT+QNETDEVSTATUS` or `AT+CGPADDR=<cid>`).

		The host network interface is not configured here; quectel-CM runs `udhcpc` for
		that, which needs root.

		Request parameters (`Ut_ParameterList`, names are case-insensitive):
		\| Name \| Meaning \| Default \|
		\|------------------\|------------------------------------------------------\|---------\|
		\| `CID` \| PDP context identifier, 1..15 \| 1 \|
		\| `APN` \| APN/DNN; when absent the profile is left untouched \| - \|
		\| `USER` \| User name \| empty \|
		\| `PASSWORD` \| Password \| empty \|
		\| `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 \|
		"""
		import logging
		import time

		log = logging.getLogger("ut.datacall")

		## Seconds between two polls of the attach state or of `AT+QNETDEVSTATUS`.
		POLL_INTERVAL = 1.0
		## Number of `AT+QNETDEVSTATUS` polls after `AT+QNETDEVCTL` before giving up (as quectel-CM).
		STATUS_POLLS = 15
		## Seconds to wait for the modem to be packet-service attached, unless `ATTACH_TIMEOUT` is given.
		DEFAULT_ATTACH_TIMEOUT = 30.0

		## `IPTYPE` parameter value to `AT+QICSGP` context type.
		IPTYPES = {"IPV4": 1, "IPV6": 2, "IPV4V6": 3}
		## `AT+QICSGP` context type to `AT+CGDCONT` PDP type.
		PDP_TYPES = {1: "IP", 2: "IPV6", 3: "IPV4V6"}
		## `AUTH` parameter aliases to `AT+QICSGP` authentication value.
		AUTHS = {"0": 0, "NONE": 0, "1": 1, "PAP": 1, "2": 2, "CHAP": 2, "3": 3, "MSCHAPV2": 3}


		class BadParameter(ValueError):
		"""!
		@brief Raised by @ref Profile.from_params when a request parameter is invalid.
		"""


		def fields(line: str) -> list[str]:
		"""!
		@brief Split the value part of an AT information response into its fields.
		@param line One response line, e.g. `'+CGPADDR: 1,"10.0.0.1"'`.
		@return The comma-separated fields after the first `:`, unquoted and stripped, e.g.
		`["1", "10.0.0.1"]`. Commas inside double quotes do not split.
		"""
		value = line.partition(":")[2]
		out, cur, quoted = [], "", False
		for ch in value:
		if ch == '"':
		quoted = not quoted
		elif ch == "," and not quoted:
		out.append(cur.strip())
		cur = ""
		else:
		cur += ch
		out.append(cur.strip())
		return out


		def has_address(addr: str) -> bool:
		"""!
		@brief Tell whether an address reported by the modem is a real, assigned address.
		@param addr An address field: dotted IPv4, IPv6, or 8 hexadecimal digits (the form
		some modems use in `+QNETDEVSTATUS`).
		@return False if @p addr is empty or all zeros (e.g. `"0.0.0.0"`, `"00000000"`, `"::"`).
		"""
		return bool(addr.strip("0.: "))


		def readable(addr: str) -> str:
		"""!
		@brief Convert an address reported by the modem to its usual text form.
		@param addr An address field, see @ref has_address.
		@return Dotted IPv4 when @p addr is 8 hexadecimal digits (stored last byte first, as
		quectel-CM decodes them: `"0100000A"` is `10.0.0.1`), @p addr unchanged otherwise.
		"""
		if len(addr) == 8 and "." not in addr and ":" not in addr:
		try:
		return ".".join(str(b) for b in reversed(bytes.fromhex(addr)))
		except ValueError:
		pass
		return addr


		class Profile:
		"""!
		@brief The validated parameters of one data call request.
		"""

		def __init__(self, cid: int, apn: str \| None, user: str, password: str, auth: int, iptype: int,
		attach_timeout: float):
		## PDP context identifier.
		self.cid = cid
		## APN/DNN, or `None` to leave the modem's profile untouched.
		self.apn = apn
		## User name (may be empty).
		self.user = user
		## Password (may be empty).
		self.password = password
		## Authentication, as `AT+QICSGP` expects it (0..3).
		self.auth = auth
		## Context type, as `AT+QICSGP` expects it (1 IPv4, 2 IPv6, 3 IPv4v6).
		self.iptype = iptype
		## Seconds to wait for attach; 0 skips the check.
		self.attach_timeout = attach_timeout

		@classmethod
		def from_params(cls, params: dict[str, str]) -> "Profile":
		"""!
		@brief Build a profile from the request parameters.
		@param params `Name -> Value`, as received; names are matched case-insensitively.
		@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()}
		try:
		cid = int(p.get("CID", "1"))
		attach_timeout = float(p.get("ATTACH_TIMEOUT", DEFAULT_ATTACH_TIMEOUT))
		except ValueError as e:
		raise BadParameter(str(e)) from None
		if not 1 <= cid <= 15:
		raise BadParameter(f"CID {cid} not in 1..15")
		if attach_timeout < 0:
		raise BadParameter(f"ATTACH_TIMEOUT {attach_timeout} is negative")
		texts = {name: p.get(name, "") for name in ("APN", "USER", "PASSWORD")}
		for name, text in texts.items():
		if '"' in text or any(ord(ch) < 0x20 for ch in text):
		raise BadParameter(f"{name} holds a quote or a control character")
		iptype = IPTYPES.get(p.get("IPTYPE", "IPV4").upper())
		if iptype is None:
		raise BadParameter(f"IPTYPE {p['IPTYPE']} not in {sorted(IPTYPES)}")
		# quectel-CM defaults to CHAP when a password is given without an auth mode
		auth = AUTHS.get(p.get("AUTH", "2" if texts["PASSWORD"] else "0").upper())
		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)


		class _At:
		"""!
		@brief Runs the AT commands of one procedure and keeps the transcript for the reply.
		"""

		def __init__(self, modem, timeout, on_written):
		self._modem = modem
		self._timeout = timeout
		self._on_written = on_written
		## Response text of every command run so far, plus any `+UT:` note.
		self.out: list[str] = []

		def __call__(self, cmd: str) -> tuple[bool, list[str]]:
		"""!
		@brief Run one AT command.
		@param cmd The AT command.
		@return `(ok, lines)`: whether the final result code was `OK`, and every response
		line received.
		"""
		ok, text = self._modem.execute(cmd, self._timeout, self._on_written)
		self._on_written = None # only the first command of the procedure matters
		self.out.append(text)
		return ok, text.split("\r\n")

		def note(self, text: str) -> None:
		"""!
		@brief Add a line of our own to the transcript, marked `+UT:`.
		"""
		log.info(text)
		self.out.append(f"+UT: {text}")

		def result(self, ok: bool) -> tuple[bool, str]:
		"""!
		@brief The procedure's `(result, result_string)`.
		"""
		return ok, "\r\n".join(self.out)


		def _info(lines: list[str], prefix: str) -> list[list[str]]:
		"""!
		@brief Fields of every response line starting with @p prefix (see @ref fields).
		"""
		return [fields(line) for line in lines if line.startswith(prefix)]


		def _set_profile(at: _At, prof: Profile) -> bool:
		"""!
		@brief Make context `prof.cid` match @p prof, writing it only when it differs.
		@return False if the modem refused both `AT+QICSGP` and `AT+CGDCONT`.
		"""
		wanted = [str(prof.iptype), prof.apn, prof.user, prof.password, str(prof.auth)]
		ok, lines = at(f"AT+QICSGP={prof.cid}")
		current = _info(lines, "+QICSGP:")
		if ok and current and current[0][:5] == wanted:
		return True
		ok, _ = at(f'AT+QICSGP={prof.cid},{prof.iptype},"{prof.apn}","{prof.user}","{prof.password}",{prof.auth}')
		if not ok:
		ok, _ = at(f'AT+CGDCONT={prof.cid},"{PDP_TYPES[prof.iptype]}","{prof.apn}"')
		return ok


		def _wait_attached(at: _At, prof: Profile) -> bool:
		"""!
		@brief Poll `AT+CGATT?` until the modem is attached or `prof.attach_timeout` elapses.
		"""
		deadline = time.monotonic() + prof.attach_timeout
		while True:
		_, lines = at("AT+CGATT?")
		state = _info(lines, "+CGATT:")
		if state and state[0][0] == "1":
		return True
		if time.monotonic() + POLL_INTERVAL > deadline:
		return False
		time.sleep(POLL_INTERVAL)


		def _context_active(at: _At, cid: int) -> bool:
		"""!
		@brief Tell whether context @p cid is active, from `AT+CGACT?`.
		"""
		_, lines = at("AT+CGACT?")
		return any(f[:2] == [str(cid), "1"] for f in _info(lines, "+CGACT:"))


		def _netdev_address(at: _At, cid: int) -> str:
		"""!
		@brief Address of the data call from `AT+QNETDEVSTATUS=<cid>`.
		@return The IPv4 address if one is assigned, else the IPv6 address, else `""`. The
		response carries IPv4 address, gateway, DHCP and two DNS, then the same for IPv6.
		"""
		ok, lines = at(f"AT+QNETDEVSTATUS={cid}")
		status = _info(lines, "+QNETDEVSTATUS")
		if not ok or not status:
		return ""
		for addr in (status[0] + [""] * 6)[0:6:5]:
		if has_address(addr):
		return readable(addr)
		return ""


		def _pdp_address(at: _At, cid: int) -> str:
		"""!
		@brief Address of context @p cid from `AT+CGPADDR=<cid>`, or `""` if none is assigned.
		"""
		_, lines = at(f"AT+CGPADDR={cid}")
		for f in _info(lines, "+CGPADDR:"):
		if f[0] == str(cid):
		for addr in f[1:]:
		if has_address(addr):
		return addr
		return ""


		def activate(modem, params: dict[str, str], timeout: float \| None = None, on_written=None) -> tuple[bool, str]:
		"""!
		@brief Set up a data call; see the @ref datacall.py "module description" for the steps.
		@param modem The modem.AtModem to run the AT commands on.
		@param params Request parameters (`Name -> Value`), see the module description.
		@param timeout Overrides the per-command default timeout of every AT command.
		@param on_written Forwarded to modem.AtModem.execute for the first AT command only.
		@return `(result, result_string)`. `result` is true only if the modem reports an IP
		address for the context. `result_string` holds the response to every AT command,
		and ends with `+UT: data call connected: cid=<cid> ip=<address>` on success or a
		`+UT:` line giving the reason on failure.
		"""
		at = _At(modem, timeout, on_written)
		try:
		prof = Profile.from_params(params)
		except BadParameter as e:
		at.note(f"bad parameter: {e}")
		return at.result(False)
		cid = prof.cid

		if prof.apn is not None and not _set_profile(at, prof):
		at.note(f"cannot set the profile of cid={cid}")
		return at.result(False)
		if prof.attach_timeout and not _wait_attached(at, prof):
		at.note(f"not attached after {prof.attach_timeout:g}s")
		return at.result(False)

		netdev_status, _ = at("AT+QNETDEVSTATUS=?")
		if netdev_status:
		dialled, _ = at(f"AT+QNETDEVCTL={cid},1,1")
		else:
		if not _context_active(at, cid):
		ok, _ = at(f"AT+CGACT=1,{cid}")
		if not ok:
		at.note(f"data call rejected: cid={cid}")
		return at.result(False)
		dialled, _ = at(f"AT+QNETDEVCTL=1,{cid},1")

		addr = ""
		if dialled and netdev_status:
		for poll in range(STATUS_POLLS):
		if poll:
		time.sleep(POLL_INTERVAL)
		addr = _netdev_address(at, cid)
		if addr:
		break
		else:
		if not dialled:
		at.note("AT+QNETDEVCTL refused: activating the context only, the host has no data path")
		if netdev_status: # otherwise the context was activated before AT+QNETDEVCTL
		ok, _ = at(f"AT+CGACT=1,{cid}")
		if not ok:
		at.note(f"data call rejected: cid={cid}")
		return at.result(False)
		addr = _pdp_address(at, cid)

		if not addr:
		at.note(f"data call failed: no address for cid={cid}")
		return at.result(False)
		at.note(f"data call connected: cid={cid} ip={addr}")
		return at.result(True)


		def deactivate(modem, params: dict[str, str], timeout: float \| None = None, on_written=None) -> tuple[bool, str]:
		"""!
		@brief Release a data call set up by @ref activate.
		@details Sends the `AT+QNETDEVCTL` disconnect matching the modem (see @ref activate), and
		falls back to `AT+CGACT=0,<cid>` if the modem refuses it.
		@param modem The modem.AtModem to run the AT commands on.
		@param params Request parameters; only `CID` is used.
		@param timeout Overrides the per-command default timeout of every AT command.
		@param on_written Forwarded to modem.AtModem.execute for the first AT command only.
		@return `(result, result_string)`, `result` being true if the modem accepted the release.
		"""
		at = _At(modem, timeout, on_written)
		try:
		cid = Profile.from_params(params).cid
		except BadParameter as e:
		at.note(f"bad parameter: {e}")
		return at.result(False)

		netdev_status, _ = at("AT+QNETDEVSTATUS=?")
		ok, _ = at(f"AT+QNETDEVCTL={cid},0,0" if netdev_status else f"AT+QNETDEVCTL=0,{cid},0")
		if not ok:
		ok, _ = at(f"AT+CGACT=0,{cid}")
		at.note(f"data call {'released' if ok else 'release failed'}: cid={cid}")
		return at.result(ok)

upper_tester/dummy_modem.py

+9 −0

Changes for upper_tester/dummy_modem.py: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -34,6 +34,15 @@ DEFAULT_SCRIPT = {
		"AT+CGATT=0": ["OK"],
		'AT+CGDCONT=1,"IPV4V6","ims"': ["OK"],
		"AT+CGACT=1,1": ["OK"],
		"AT+CGACT=0,1": ["OK"],
		"AT+CGACT?": ["+CGACT: 1,1", "OK"],
		"AT+CGATT?": ["+CGATT: 1", "OK"],
		"AT+QICSGP=1": ['+QICSGP: 1,"internet","","",0', "OK"],
		'AT+QICSGP=1,1,"internet","","",0': ["OK"],
		"AT+QNETDEVSTATUS=?": ["+QNETDEVSTATUS: (1-15)", "OK"],
		"AT+QNETDEVCTL=1,1,1": ["OK"],
		"AT+QNETDEVCTL=1,0,0": ["OK"],
		"AT+QNETDEVSTATUS=1": ['+QNETDEVSTATUS: "10.0.0.1","10.0.0.2","10.0.0.2","8.8.8.8","8.8.4.4"', "OK"],
		"AT+CGPADDR=1": ['+CGPADDR: 1,"10.0.0.1"', "OK"],
		'AT+CDU=1,"tel:0123456789"': ["OK"],
		"AT+CHCCS=1": ["OK"],

upper_tester/engine.py

+7 −2

Changes for upper_tester/engine.py: 7 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -4,7 +4,7 @@
		"""
		import logging

		from .mmi import UnsupportedMmi, steps_for
		from .mmi import MMI_PROCEDURES, UnsupportedMmi, steps_for

		log = logging.getLogger("ut.engine")

		@@ -31,7 +31,9 @@ class Engine:
		"""!
		@brief Execute one decoded request. Blocking; call from a worker thread.
		@details For `type: "AT"`, @p req["cmd"] is sent to the modem as-is. For
		`type: "MMI"`, @p req["cmd"] is resolved via mmi.steps_for into one or
		`type: "MMI"`, @p req["cmd"] is run by its procedure if it has one in
		mmi.MMI_PROCEDURES (e.g. `ACTIVATE_BEARER`, see datacall.activate);
		otherwise it is resolved via mmi.steps_for into one or
		more AT commands, executed in order; the sequence stops at the first
		step that fails, and @p on_written only fires for the first step (an
		MMI command is one logical operation from the test system's point of
		@@ -59,6 +61,9 @@ class Engine:
		return self.modem.execute(cmd, timeout, on_written)
		params = {p["name"]: p["value"] for p in req.get("params") or []}
		log.debug("dispatch: MMI %s params=%s", cmd, params)
		procedure = MMI_PROCEDURES.get(cmd.upper())
		if procedure:
		return procedure(self.modem, params, timeout, on_written)
		try:
		steps = steps_for(cmd, params)
		except UnsupportedMmi:

upper_tester/mmi.py

+11 −0

Changes for upper_tester/mmi.py: 11 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -9,6 +9,8 @@
		"""
		import logging

		from . import datacall

		log = logging.getLogger("ut.mmi")

		## Table of MMI command name (as sent by the TITAN test system, upper-cased) to the list of
		@@ -25,6 +27,15 @@ MMI_TABLE = {
		"RELEASE_CALL": ["AT+CHUP"],
		}

		## Table of MMI command name to the procedure that implements it, for commands that need
		## more than a fixed list of AT commands (decisions, polling). A procedure is a
		## `callable(modem, params, timeout, on_written) -> (result, result_string)`; looked up
		## before @ref MMI_TABLE (see engine.Engine.run).
		MMI_PROCEDURES = {
		"ACTIVATE_BEARER": datacall.activate,
		"DEACTIVATE_BEARER": datacall.deactivate,
		}


		class UnsupportedMmi(KeyError):
		"""!