Skip to content

feat: tmq supports token #385

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zitsen merged 8 commits into from
Mar 12, 2026
Merged

feat: tmq supports token #385

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6b2b80e
feat(ws): tmq supports token
qevolg Mar 4, 2026
788cdbb
fix: fix aws-lc-rs vuln
qevolg Mar 5, 2026
e5f92c4
feat: native tmq supports token
qevolg Mar 5, 2026
30ee18a
test: update test case
qevolg Mar 5, 2026
5f13622
test: modify test case
qevolg Mar 5, 2026
3986927
chore: update version to 2.8.9 and add changelog entries for token au…
qevolg Mar 5, 2026
b3309c6
chore(ws): update taos version
qevolg Mar 9, 2026
ccbea1b
refactor(native): rename td.connect.token to td.connect.bearer_token
qevolg Mar 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions taos-ws-py/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 2 additions & 2 deletions taos-ws-py/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,13 @@ version = "0.17.3"
features = ["extension-module", "anyhow", "chrono", "abi3-py37"]

[target.'cfg(windows)'.dependencies]
taos = { git = "https://github.com/taosdata/taos-connector-rust.git", branch = "main", default-features = false, version = "0.12.3", features = [
taos = { git = "https://github.com/taosdata/taos-connector-rust.git", branch = "feat/main/6735261785", default-features = false, version = "0.12.3", features = [
"optin",
"ws-rustls",
"ws-rustls-aws-lc-crypto-provider",
] }
[target.'cfg(unix)'.dependencies]
taos = { git = "https://github.com/taosdata/taos-connector-rust.git", branch = "main", default-features = false, version = "0.12.3", features = [
taos = { git = "https://github.com/taosdata/taos-connector-rust.git", branch = "feat/main/6735261785", default-features = false, version = "0.12.3", features = [
"optin",
"ws-rustls",
"ws-rustls-ring-crypto-provider",
Expand Down
11 changes: 11 additions & 0 deletions taos-ws-py/src/consumer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,9 @@ impl Consumer {
}

let mut tz = dsn_cfg.get("timezone").cloned();
if let Some(token) = dsn_cfg.remove("td.connect.bearer_token") {
dsn_cfg.set("bearer_token", token);
}

if let Some(args) = conf {
if let Some(scheme) = args
Expand Down Expand Up @@ -105,6 +108,12 @@ impl Consumer {
if let Some(value) = args.get_item("td.connect.token").or(args.get_item("token")) {
dsn_cfg.set("token", value.extract::<String>()?);
}
if let Some(value) = args
.get_item("td.connect.bearer_token")
.or(args.get_item("bearer_token"))
{
dsn_cfg.set("bearer_token", value.extract::<String>()?);
}
if let Some(value) = args.get_item("timezone") {
tz = Some(value.extract::<String>()?);
}
Expand Down Expand Up @@ -135,6 +144,8 @@ impl Consumer {
"td.connect.user",
"td.connect.pass",
"group.id",
"bearer_token",
"td.connect.bearer_token",
];

for (key, value) in args {
Expand Down
125 changes: 123 additions & 2 deletions taos-ws-py/tests/test_tmq.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
import pytest
import taosws
import time
import utils
from taosws import Consumer


Expand Down Expand Up @@ -55,7 +58,6 @@ def setup():
"alter table `table` drop column new1",
]
for statement in statements:
# print(statement)
cursor.execute(statement)


Expand All @@ -69,7 +71,6 @@ def test_tmq():
"min.poll.rows": "129",
}
consumer = Consumer(conf)

consumer.subscribe(["ws_tmq_meta"])

while 1:
Expand All @@ -95,6 +96,126 @@ def test_tmq():
consumer.unsubscribe()


@pytest.mark.skipif(not utils.TEST_TD_ENTERPRISE, reason="only for TDengine Enterprise")
def test_tmq_with_token():
conn = taosws.connect("ws://localhost:6041")
conn.execute("drop token if exists token_1772607422")
conn.execute("drop topic if exists topic_1772607422")
conn.execute("drop database if exists test_1772607422")
conn.execute("create database test_1772607422")
conn.execute("create topic topic_1772607422 as database test_1772607422")

rs = conn.query(f"create token token_1772607422 from user {utils.test_username()}")
token = next(iter(rs))[0]

consumer = Consumer(conf={
"td.connect.websocket.scheme": "ws",
"td.connect.ip": "localhost",
"td.connect.port": 6041,
"td.connect.user": "invalid_user",
"td.connect.pass": "invalid_pass",
"td.connect.bearer_token": token,
"group.id": "1001",
"client.id": "1001",
})
consumer.subscribe(["topic_1772607422"])
consumer.unsubscribe()

consumer2 = Consumer(conf={
"td.connect.websocket.scheme": "ws",
"td.connect.ip": "localhost",
"td.connect.port": 6041,
"td.connect.user": "invalid_user",
"td.connect.pass": "invalid_pass",
"bearer_token": token,
"group.id": "1001",
"client.id": "1001",
})
consumer2.subscribe(["topic_1772607422"])
consumer2.unsubscribe()

consumer3 = Consumer(conf={
"td.connect.websocket.scheme": "ws",
"td.connect.ip": "localhost",
"td.connect.port": 6041,
"td.connect.user": "invalid_user",
"td.connect.pass": "invalid_pass",
"td.connect.bearer_token": token,
"bearer_token": "invalid_token",
"group.id": "1001",
"client.id": "1001",
})
consumer3.subscribe(["topic_1772607422"])
consumer3.unsubscribe()

consumer4 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token={token}")
consumer4.subscribe(["topic_1772607422"])
consumer4.unsubscribe()

consumer5 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&bearer_token={token}")
consumer5.subscribe(["topic_1772607422"])
consumer5.unsubscribe()

consumer6 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token={token}&bearer_token=invalid_token")
consumer6.subscribe(["topic_1772607422"])
consumer6.unsubscribe()

time.sleep(3)
conn.execute("drop token if exists token_1772607422")
conn.execute("drop topic if exists topic_1772607422")
conn.execute("drop database if exists test_1772607422")
conn.close()


@pytest.mark.skipif(not utils.TEST_TD_ENTERPRISE, reason="only for TDengine Enterprise")
def test_tmq_with_invalid_token():
conn = taosws.connect("ws://localhost:6041")
try:
conn.execute("drop topic if exists topic_1772611547")
conn.execute("drop database if exists test_1772611547")
conn.execute("create database test_1772611547")
conn.execute("create topic topic_1772611547 as database test_1772611547")

consumer = Consumer(conf={
"td.connect.websocket.scheme": "ws",
"td.connect.ip": "localhost",
"td.connect.port": 6041,
"td.connect.user": "invalid_user",
"td.connect.pass": "invalid_pass",
"td.connect.bearer_token": "invalid_token",
"group.id": "1001",
"client.id": "1001",
})
with pytest.raises(Exception, match=r"init tscObj with token failed"):
consumer.subscribe(["topic_1772611547"])

consumer2 = Consumer(conf={
"td.connect.websocket.scheme": "ws",
"td.connect.ip": "localhost",
"td.connect.port": 6041,
"td.connect.user": "invalid_user",
"td.connect.pass": "invalid_pass",
"bearer_token": "invalid_token",
"group.id": "1001",
"client.id": "1001",
})
with pytest.raises(Exception, match=r"init tscObj with token failed"):
consumer2.subscribe(["topic_1772611547"])

consumer3 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&td.connect.bearer_token=invalid_token")
with pytest.raises(Exception, match=r"init tscObj with token failed"):
consumer3.subscribe(["topic_1772611547"])

consumer4 = Consumer(dsn = f"ws://invalid_user:invalid_pass@localhost:6041?group.id=1001&client.id=1001&bearer_token=invalid_token")
with pytest.raises(Exception, match=r"init tscObj with token failed"):
consumer4.subscribe(["topic_1772611547"])
finally:
time.sleep(3)
conn.execute("drop topic if exists topic_1772611547")
conn.execute("drop database if exists test_1772611547")
conn.close()


def show_env():
import os

Expand Down
Loading