Enhance UT sequence (AT+COPTS=2 replaced by AT+CFUN=1) (57bfc337) · Commits · INT - Core Network and Interoperability Testing / 5G Core / NAS

ttcn/LibNGAP/lib_system/LibNGAP_Functions.ttcn

+13 −0

Changes for ttcn/LibNGAP/lib_system/LibNGAP_Functions.ttcn: 13 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -355,6 +355,19 @@ module LibNGAP_Functions {
		));
		}
		f_NASPDU_Get(vc_recvNGAP_PDU);
		// A UE holding a 5G NAS security context from a previous registration sends its Registration
		// Request integrity protected (TS 24.501 clause 4.4.6): unwrap it. That context is unknown
		// here, so the MAC is not checked; the authentication that follows establishes a new one
		if (ischosen(vc_recvNgNasUl_Msg.security_Protected_Nas_Message)) {
		if (vc_recvNgNasUl_Msg.security_Protected_Nas_Message.securityHeaderType == tsc_SHT_IntegrityProtected) {
		var bitstring v_plain := oct2bit(vc_recvNgNasUl_Msg.security_Protected_Nas_Message.plainNASMessage);
		var NG_NAS_UL_Message_Type v_inner;
		if (decvalue(v_plain, v_inner) == 0) {
		log("f_await_registration_request: integrity protected NAS message unwrapped: ", v_inner);
		vc_recvNgNasUl_Msg := v_inner;
		}
		}
		}
		if (match(vc_recvNgNasUl_Msg, mw_NG_REGISTRATION_REQUEST)) {
		log("Received NG_REGISTRATION_REQUEST NAS message");

upper_tester/README.md

+6 −4

Changes for upper_tester/README.md: 6 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -47,15 +47,17 @@ call is reported once and never redialled. See `datacall.py`.
		\| `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 \|

		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).
		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
		(`quectel-CM` runs `udhcpc` for that, as root).

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

upper_tester/datacall.py

+42 −26

Changes for upper_tester/datacall.py: 42 added lines, 26 removed lines.
Original line number	Diff line number	Diff line
		@@ -9,19 +9,22 @@
		-# 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:
		-# Dial: `AT+CGACT=1,<cid>` (unless `AT+CGACT?` shows the context already active).
		The network then sees the session (PDU Session Establishment), but the host gets
		no data path through the modem's network interface.
		-# Address: the call counts as connected only once the modem reports an IP
		address (`AT+CGPADDR=<cid>`).

		`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:
		- `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>`).
		`AT+QNETDEVCTL=1,<cid>,1`, and the address is read with `AT+CGPADDR=<cid>`;
		- `AT+QNETDEVCTL` itself is refused: as without `NETDEV=1`, and the reply says so.

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

		Request parameters (`Ut_ParameterList`, names are case-insensitive):
		@@ -34,6 +37,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 \|
		"""
		import logging
		import time
		@@ -113,7 +117,7 @@ class Profile:
		"""

		def __init__(self, cid: int, apn: str \| None, user: str, password: str, auth: int, iptype: int,
		attach_timeout: float):
		attach_timeout: float, netdev: bool = False):
		## PDP context identifier.
		self.cid = cid
		## APN/DNN, or `None` to leave the modem's profile untouched.
		@@ -128,6 +132,8 @@ class Profile:
		self.iptype = iptype
		## Seconds to wait for attach; 0 skips the check.
		self.attach_timeout = attach_timeout
		## Whether to dial and release with `AT+QNETDEVCTL` instead of `AT+CGACT` alone.
		self.netdev = netdev

		@classmethod
		def from_params(cls, params: dict[str, str]) -> "Profile":
		@@ -148,6 +154,8 @@ 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"):
		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():
		if '"' in text or any(ord(ch) < 0x20 for ch in text):
		@@ -159,7 +167,8 @@ class Profile:
		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)
		return cls(cid, p.get("APN"), texts["USER"], texts["PASSWORD"], auth, iptype, attach_timeout,
		p.get("NETDEV", "0") == "1")


		class _At:
		@@ -302,18 +311,24 @@ def activate(modem, params: dict[str, str], timeout: float \| None = None, on_wri
		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):
		def activate_context() -> bool:
		if _context_active(at, cid):
		return True
		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")
		return ok

		addr = ""
		if not prof.netdev:
		if not activate_context():
		return at.result(False)
		addr = _pdp_address(at, cid)
		else:
		netdev_status, _ = at("AT+QNETDEVSTATUS=?")
		if not netdev_status and not activate_context():
		return at.result(False)
		dialled, _ = at(f"AT+QNETDEVCTL={cid},1,1" if netdev_status else f"AT+QNETDEVCTL=1,{cid},1")
		if dialled and netdev_status:
		for poll in range(STATUS_POLLS):
		if poll:
		@@ -324,10 +339,7 @@ def activate(modem, params: dict[str, str], timeout: float \| None = None, on_wri
		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}")
		if netdev_status and not activate_context():
		return at.result(False)
		addr = _pdp_address(at, cid)

		@@ -341,21 +353,25 @@ def activate(modem, params: dict[str, str], timeout: float \| None = None, on_wri
		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.
		@details Sends `AT+CGACT=0,<cid>`. With `NETDEV=1`, first sends the `AT+QNETDEVCTL`
		disconnect matching the modem (see @ref activate), and only 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 params Request parameters; only `CID` and `NETDEV` are 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
		prof = Profile.from_params(params)
		except BadParameter as e:
		at.note(f"bad parameter: {e}")
		return at.result(False)
		cid = prof.cid

		ok = False
		if prof.netdev:
		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:

upper_tester/tests/test_ut.py

+32 −12

Changes for upper_tester/tests/test_ut.py: 32 added lines, 12 removed lines.
Original line number	Diff line number	Diff line
		@@ -155,7 +155,8 @@ CGACT_SCRIPT = {

		class TestDataCall(Base):
		"""! @brief ACTIVATE_BEARER / DEACTIVATE_BEARER (datacall.py) through the NDJSON
		protocol, for each of the three ways a data call can be dialled, and its failures. """
		protocol: with AT+CGACT alone (the default), with AT+QNETDEVCTL (NETDEV=1) in each of
		the ways it can be dialled, and the failures. """

		async def asyncSetUp(self):
		await super().asyncSetUp()
		@@ -167,9 +168,27 @@ class TestDataCall(Base):
		plist = [{"name": k, "value": v} for k, v in params.items()]
		return await self.call(id=1, type="MMI", cmd=cmd, params=plist)

		async def test_default_dials_with_cgact_only(self):
		self.fake.script.update(CGACT_SCRIPT)
		r = await self.mmi("ACTIVATE_BEARER")
		self.assertTrue(r["result"], r)
		self.assertIn("+UT: data call connected: cid=1 ip=10.0.0.1", r["result_string"])
		self.assertEqual(self.fake.received, ["AT+CGATT?", "AT+CGACT?", "AT+CGACT=1,1", "AT+CGPADDR=1"])

		async def test_default_skips_active_context(self):
		self.fake.script.update(CGACT_SCRIPT)
		self.fake.script["AT+CGACT?"] = ["+CGACT: 1,1", "OK"]
		self.assertTrue((await self.mmi("ACTIVATE_BEARER"))["result"])
		self.assertNotIn("AT+CGACT=1,1", self.fake.received)

		async def test_default_deactivate(self):
		self.fake.script.update(CGACT_SCRIPT)
		self.assertTrue((await self.mmi("DEACTIVATE_BEARER", CID="1"))["result"])
		self.assertEqual(self.fake.received, ["AT+CGACT=0,1"])

		async def test_netdev_style_sets_profile_and_connects(self):
		self.fake.script.update(NETDEV_SCRIPT)
		r = await self.mmi("ACTIVATE_BEARER", APN="internet")
		r = await self.mmi("ACTIVATE_BEARER", APN="internet", NETDEV="1")
		self.assertTrue(r["result"], r)
		self.assertIn("+UT: data call connected: cid=1 ip=10.0.0.1", r["result_string"])
		self.assertEqual(self.fake.received, [
		@@ -179,8 +198,8 @@ class TestDataCall(Base):

		async def test_profile_untouched_when_same_or_no_apn(self):
		self.fake.script.update(NETDEV_SCRIPT)
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", APN="old"))["result"])
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", Cid="1"))["result"])
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", APN="old", NETDEV="1"))["result"])
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", Cid="1", NETDEV="1"))["result"])
		self.assertEqual(self.fake.received.count("AT+QICSGP=1"), 1)
		self.assertFalse([c for c in self.fake.received if c.startswith("AT+QICSGP=1,")])

		@@ -188,11 +207,11 @@ class TestDataCall(Base):
		self.fake.script.update(NETDEV_SCRIPT)
		del self.fake.script['AT+QICSGP=1,1,"internet","","",0']
		self.fake.script['AT+CGDCONT=1,"IP","internet"'] = ["OK"]
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", APN="internet"))["result"])
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", APN="internet", NETDEV="1"))["result"])

		async def test_cgact_style_connects(self):
		self.fake.script.update(CGACT_SCRIPT)
		r = await self.mmi("ACTIVATE_BEARER")
		r = await self.mmi("ACTIVATE_BEARER", NETDEV="1")
		self.assertTrue(r["result"], r)
		self.assertIn("ip=10.0.0.1", r["result_string"])
		self.assertEqual(self.fake.received, [
		@@ -203,13 +222,13 @@ class TestDataCall(Base):
		async def test_cgact_style_skips_active_context(self):
		self.fake.script.update(CGACT_SCRIPT)
		self.fake.script["AT+CGACT?"] = ["+CGACT: 1,1", "OK"]
		self.assertTrue((await self.mmi("ACTIVATE_BEARER"))["result"])
		self.assertTrue((await self.mmi("ACTIVATE_BEARER", NETDEV="1"))["result"])
		self.assertNotIn("AT+CGACT=1,1", self.fake.received)

		async def test_falls_back_to_context_only_without_netdev(self):
		self.fake.script.update(CGACT_SCRIPT)
		del self.fake.script["AT+QNETDEVCTL=1,1,1"]
		r = await self.mmi("ACTIVATE_BEARER")
		r = await self.mmi("ACTIVATE_BEARER", NETDEV="1")
		self.assertTrue(r["result"], r)
		self.assertIn("the host has no data path", r["result_string"])

		@@ -224,7 +243,7 @@ class TestDataCall(Base):
		async def test_no_address_fails(self):
		self.fake.script.update(NETDEV_SCRIPT)
		self.fake.script["AT+QNETDEVSTATUS=1"] = ['+QNETDEVSTATUS: "0.0.0.0","0.0.0.0"', "OK"]
		r = await self.mmi("ACTIVATE_BEARER")
		r = await self.mmi("ACTIVATE_BEARER", NETDEV="1")
		self.assertFalse(r["result"])
		self.assertEqual(self.fake.received.count("AT+QNETDEVSTATUS=1"), datacall.STATUS_POLLS)

		@@ -241,7 +260,8 @@ class TestDataCall(Base):
		self.assertNotIn("AT+CGATT?", self.fake.received)

		async def test_bad_parameter(self):
		for params in ({"CID": "0"}, {"CID": "x"}, {"APN": 'a"b'}, {"IPTYPE": "IPX"}, {"AUTH": "9"}):
		for params in ({"CID": "0"}, {"CID": "x"}, {"APN": 'a"b'}, {"IPTYPE": "IPX"}, {"AUTH": "9"},
		{"NETDEV": "2"}):
		r = await self.mmi("ACTIVATE_BEARER", **params)
		self.assertFalse(r["result"], params)
		self.assertIn("+UT: bad parameter", r["result_string"])
		@@ -249,13 +269,13 @@ class TestDataCall(Base):

		async def test_deactivate(self):
		self.fake.script.update(NETDEV_SCRIPT)
		self.assertTrue((await self.mmi("DEACTIVATE_BEARER"))["result"])
		self.assertTrue((await self.mmi("DEACTIVATE_BEARER", NETDEV="1"))["result"])
		self.assertEqual(self.fake.received[-1], "AT+QNETDEVCTL=1,0,0")

		async def test_deactivate_falls_back_to_cgact(self):
		self.fake.script.update(CGACT_SCRIPT)
		del self.fake.script["AT+QNETDEVCTL=0,1,0"]
		self.assertTrue((await self.mmi("DEACTIVATE_BEARER", CID="1"))["result"])
		self.assertTrue((await self.mmi("DEACTIVATE_BEARER", CID="1", NETDEV="1"))["result"])
		self.assertEqual(self.fake.received[-2:], ["AT+QNETDEVCTL=0,1,0", "AT+CGACT=0,1"])

		def test_address_helpers(self):