+ def tdls_link_status(self, peer):
+ cmd = "TDLS_LINK_STATUS " + peer
+ ret = self.group_request(cmd)
+ if "FAIL" in ret:
+ raise Exception("Failed to request TDLS link status")
+ return ret
+
+ def tspecs(self):
+ """Return (tsid, up) tuples representing current tspecs"""
+ res = self.request("WMM_AC_STATUS")
+ tspecs = re.findall(r"TSID=(\d+) UP=(\d+)", res)
+ tspecs = [tuple(map(int, tspec)) for tspec in tspecs]
+
+ logger.debug("tspecs: " + str(tspecs))
+ return tspecs
+
+ def add_ts(self, tsid, up, direction="downlink", expect_failure=False,
+ extra=None):
+ params = {
+ "sba": 9000,
+ "nominal_msdu_size": 1500,
+ "min_phy_rate": 6000000,
+ "mean_data_rate": 1500,
+ }
+ cmd = "WMM_AC_ADDTS %s tsid=%d up=%d" % (direction, tsid, up)
+ for (key, value) in params.iteritems():
+ cmd += " %s=%d" % (key, value)
+ if extra:
+ cmd += " " + extra
+
+ if self.request(cmd).strip() != "OK":
+ raise Exception("ADDTS failed (tsid=%d up=%d)" % (tsid, up))
+
+ if expect_failure:
+ ev = self.wait_event(["TSPEC-REQ-FAILED"], timeout=2)
+ if ev is None:
+ raise Exception("ADDTS failed (time out while waiting failure)")
+ if "tsid=%d" % (tsid) not in ev:
+ raise Exception("ADDTS failed (invalid tsid in TSPEC-REQ-FAILED")
+ return
+
+ ev = self.wait_event(["TSPEC-ADDED"], timeout=1)
+ if ev is None:
+ raise Exception("ADDTS failed (time out)")
+ if "tsid=%d" % (tsid) not in ev:
+ raise Exception("ADDTS failed (invalid tsid in TSPEC-ADDED)")
+
+ if not (tsid, up) in self.tspecs():
+ raise Exception("ADDTS failed (tsid not in tspec list)")
+
+ def del_ts(self, tsid):
+ if self.request("WMM_AC_DELTS %d" % (tsid)).strip() != "OK":
+ raise Exception("DELTS failed")
+
+ ev = self.wait_event(["TSPEC-REMOVED"], timeout=1)
+ if ev is None:
+ raise Exception("DELTS failed (time out)")
+ if "tsid=%d" % (tsid) not in ev:
+ raise Exception("DELTS failed (invalid tsid in TSPEC-REMOVED)")
+
+ tspecs = [(t, u) for (t, u) in self.tspecs() if t == tsid]
+ if tspecs:
+ raise Exception("DELTS failed (still in tspec list)")
+