From 0644fb630b0fd0b8008482788c3f20dc5a626b34 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 11:29:10 +0000 Subject: [PATCH 1/6] fix(security): restrict permissions of auth token file to 0600 - Use `os.open` with `0o600` when creating the token file to ensure it is not readable by others. - Explicitly `os.chmod` the file to `0o600` to secure existing files. - Add `tests/test_auth_security.py` to verify the fix and ensure no regressions. - Mock `os.path.expanduser` in tests to prevent side effects on user's home directory. Co-authored-by: refraction-ray <35157286+refraction-ray@users.noreply.github.com> --- tensorcircuit/cloud/apis.py | 4 ++- tests/test_auth_security.py | 63 +++++++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 tests/test_auth_security.py diff --git a/tensorcircuit/cloud/apis.py b/tensorcircuit/cloud/apis.py index 7328f21d..faf07b2f 100644 --- a/tensorcircuit/cloud/apis.py +++ b/tensorcircuit/cloud/apis.py @@ -234,7 +234,9 @@ def set_token( if cached: file_token = {k: b64encode_s(v) for k, v in saved_token.items()} if file_token: - with open(authpath, "w") as f: + fd = os.open(authpath, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + os.chmod(authpath, 0o600) json.dump(file_token, f) return saved_token diff --git a/tests/test_auth_security.py b/tests/test_auth_security.py new file mode 100644 index 00000000..69530f4b --- /dev/null +++ b/tests/test_auth_security.py @@ -0,0 +1,63 @@ +import os +import stat +import json +import pytest +from unittest.mock import patch +from tensorcircuit.cloud.apis import set_token, saved_token + +def test_token_file_permissions(tmp_path): + # Mock os.path.expanduser to return tmp_path + # We patch where it is used in tensorcircuit.cloud.apis + with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): + authpath = tmp_path / ".tc.auth.json" + + # Ensure clean state for saved_token global variable if necessary + # set_token with clear=True clears the global saved_token + set_token(clear=True) + + # Scenario 1: File creation (new file) + # Set a dummy token + set_token(token="dummy_token_1", provider="tencent", cached=True) + + # Verify file exists + assert authpath.exists() + + # Verify permissions (only on POSIX systems where these bits are meaningful) + if os.name == "posix": + st = os.stat(authpath) + # Check that group and others have no permissions (should be 0) + assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Scenario 2: File update (existing file with insecure permissions) + if os.name == "posix": + # Manually set insecure permissions to simulate an existing insecure file + os.chmod(authpath, 0o666) + st_before = os.stat(authpath) + # Verify it is indeed insecure (readable by others) + assert (st_before.st_mode & stat.S_IROTH) + + # Update token (add another token or update existing) + # This triggers the write logic again + set_token(token="dummy_token_2", provider="local", cached=True) + + # Verify permissions again (should be fixed to 0600) + if os.name == "posix": + st_after = os.stat(authpath) + assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Verify content updated and readable + with open(authpath, "r") as f: + data = json.load(f) + assert data is not None + # Check if both tokens are present (implementation detail: set_token updates the dict) + # Keys are typically "provider::device" or "provider::" + # Since we added "tencent" and "local", we expect keys for them. + # But the exact key format depends on set_token logic. + # We are mainly testing permissions here. + +if __name__ == "__main__": + # Manually running with pytest main if executed as script + import sys + sys.exit(pytest.main([__file__])) From a9ce712a3d102a3a51a53737833222ba7e87d829 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 13:11:24 +0000 Subject: [PATCH 2/6] refactor: move security test to tests/test_cloud.py and remove separate test file - Removed `tests/test_auth_security.py`. - Moved `test_token_file_permissions` to `tests/test_cloud.py`. - Refactored `tests/test_cloud.py` to use per-test skipping logic instead of module-level skip, allowing unit tests like `test_token_file_permissions` to run without `TC_CLOUD_TEST` env var. Co-authored-by: refraction-ray <35157286+refraction-ray@users.noreply.github.com> --- tests/test_auth_security.py | 63 -------------------------------- tests/test_cloud.py | 71 +++++++++++++++++++++++++++++++++++-- 2 files changed, 68 insertions(+), 66 deletions(-) delete mode 100644 tests/test_auth_security.py diff --git a/tests/test_auth_security.py b/tests/test_auth_security.py deleted file mode 100644 index 69530f4b..00000000 --- a/tests/test_auth_security.py +++ /dev/null @@ -1,63 +0,0 @@ -import os -import stat -import json -import pytest -from unittest.mock import patch -from tensorcircuit.cloud.apis import set_token, saved_token - -def test_token_file_permissions(tmp_path): - # Mock os.path.expanduser to return tmp_path - # We patch where it is used in tensorcircuit.cloud.apis - with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): - authpath = tmp_path / ".tc.auth.json" - - # Ensure clean state for saved_token global variable if necessary - # set_token with clear=True clears the global saved_token - set_token(clear=True) - - # Scenario 1: File creation (new file) - # Set a dummy token - set_token(token="dummy_token_1", provider="tencent", cached=True) - - # Verify file exists - assert authpath.exists() - - # Verify permissions (only on POSIX systems where these bits are meaningful) - if os.name == "posix": - st = os.stat(authpath) - # Check that group and others have no permissions (should be 0) - assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 - - # Scenario 2: File update (existing file with insecure permissions) - if os.name == "posix": - # Manually set insecure permissions to simulate an existing insecure file - os.chmod(authpath, 0o666) - st_before = os.stat(authpath) - # Verify it is indeed insecure (readable by others) - assert (st_before.st_mode & stat.S_IROTH) - - # Update token (add another token or update existing) - # This triggers the write logic again - set_token(token="dummy_token_2", provider="local", cached=True) - - # Verify permissions again (should be fixed to 0600) - if os.name == "posix": - st_after = os.stat(authpath) - assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 - - # Verify content updated and readable - with open(authpath, "r") as f: - data = json.load(f) - assert data is not None - # Check if both tokens are present (implementation detail: set_token updates the dict) - # Keys are typically "provider::device" or "provider::" - # Since we added "tencent" and "local", we expect keys for them. - # But the exact key format depends on set_token logic. - # We are mainly testing permissions here. - -if __name__ == "__main__": - # Manually running with pytest main if executed as script - import sys - sys.exit(pytest.main([__file__])) diff --git a/tests/test_cloud.py b/tests/test_cloud.py index fb24d318..17dae8cd 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -3,6 +3,9 @@ import time import pytest import numpy as np +import stat +import json +from unittest.mock import patch thisfile = os.path.abspath(__file__) modulepath = os.path.dirname(os.path.dirname(thisfile)) @@ -12,11 +15,10 @@ from tensorcircuit.cloud import apis, wrapper from tensorcircuit.results import counts -if "TC_CLOUD_TEST" not in os.environ: - pytest.skip(allow_module_level=True) - # skip on CI due to no token +skip_cloud = pytest.mark.skipif("TC_CLOUD_TEST" not in os.environ, reason="no token") +@skip_cloud def test_get_token(): print(apis.get_token(provider="Tencent")) p = apis.get_provider("tencent") @@ -24,6 +26,7 @@ def test_get_token(): print(p.get_device("simulator:tc").get_token()) +@skip_cloud def test_list_devices(): print(apis.list_devices()) p = apis.get_provider() @@ -31,6 +34,7 @@ def test_list_devices(): print(p.list_devices(state="on")) +@skip_cloud def test_get_device(): d1 = apis.get_device(device="tencent::hello") assert d1.name == "hello" @@ -47,6 +51,7 @@ def test_get_device(): assert d4.provider.name == "tencent" +@skip_cloud def test_get_device_cache(): d1 = apis.get_device("local::testing") d2 = apis.get_device(provider="local", device="testing") @@ -60,6 +65,7 @@ def test_get_device_cache(): assert id(d4) != id(d1) +@skip_cloud def test_list_properties(): d = apis.get_device(device="simulator:aer") print(d.list_properties()) @@ -68,6 +74,7 @@ def test_list_properties(): apis.list_properties(device="hell") +@skip_cloud def test_submit_task(): c = tc.Circuit(3) c.H(0) @@ -80,6 +87,7 @@ def test_submit_task(): assert t.get_logical_physical_mapping() == {0: 0, 1: 1, 2: 2} +@skip_cloud def test_resubmit_task(): c = tc.Circuit(3) c.H(0) @@ -91,6 +99,7 @@ def test_resubmit_task(): print(t1.details(wait=True)) +@skip_cloud def test_get_task(): apis.set_device("simulator:tcn1") c = tc.Circuit(2) @@ -104,17 +113,20 @@ def test_get_task(): apis.set_device() +@skip_cloud def test_list_tasks(): d = apis.get_device(device="simulator:aer") print(d.list_tasks()) print(apis.list_tasks(device="simulator:tc")) +@skip_cloud def test_local_list_device(): dd = apis.list_devices(provider="local") assert dd[0].name == "testing" +@skip_cloud def test_local_submit_task(): c = tc.Circuit(2) c.h(0) @@ -127,10 +139,12 @@ def test_local_submit_task(): print(t.get_device()) +@skip_cloud def test_local_list_tasks(): print(apis.list_tasks(provider="local")) +@skip_cloud def test_local_batch_submit(): apis.set_provider("local") c = tc.Circuit(2) @@ -147,6 +161,7 @@ def test_local_batch_submit(): apis.set_provider("tencent") +@skip_cloud def test_batch_exp_ps(): pss = [[1, 0], [0, 3]] c = tc.Circuit(2) @@ -168,6 +183,7 @@ def test_batch_exp_ps(): ) +@skip_cloud def test_batch_submit_template(): run = tc.cloud.wrapper.batch_submit_template( device="simulator:tc", batch_limit=2, prior=10 @@ -184,6 +200,7 @@ def test_batch_submit_template(): assert len(rs) == 4 +@skip_cloud def test_allz_batch(): n = 5 @@ -217,3 +234,51 @@ def qmlf(inputs, params, device=None): params = np.ones([2, n]) print(qmlf(inputs, params, device="9gmon")) print(qmlf(inputs, params)) + + +def test_token_file_permissions(tmp_path): + # Mock os.path.expanduser to return tmp_path + # We patch where it is used in tensorcircuit.cloud.apis + with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): + authpath = tmp_path / ".tc.auth.json" + + # Ensure clean state for saved_token global variable if necessary + # set_token with clear=True clears the global saved_token + apis.set_token(clear=True) + + # Scenario 1: File creation (new file) + # Set a dummy token + apis.set_token(token="dummy_token_1", provider="tencent", cached=True) + + # Verify file exists + assert authpath.exists() + + # Verify permissions (only on POSIX systems where these bits are meaningful) + if os.name == "posix": + st = os.stat(authpath) + # Check that group and others have no permissions (should be 0) + assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Scenario 2: File update (existing file with insecure permissions) + if os.name == "posix": + # Manually set insecure permissions to simulate an existing insecure file + os.chmod(authpath, 0o666) + st_before = os.stat(authpath) + # Verify it is indeed insecure (readable by others) + assert (st_before.st_mode & stat.S_IROTH) + + # Update token (add another token or update existing) + # This triggers the write logic again + apis.set_token(token="dummy_token_2", provider="local", cached=True) + + # Verify permissions again (should be fixed to 0600) + if os.name == "posix": + st_after = os.stat(authpath) + assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Verify content updated and readable + with open(authpath, "r") as f: + data = json.load(f) + assert data is not None From e75ce0df8d992737bf22a0e61f0d3fd04b4bcf29 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 14:38:47 +0000 Subject: [PATCH 3/6] refactor: split tests/test_cloud.py into skipped and non-skipped classes - Created `TestCloud` class decorated with `@skip_cloud` for integration tests requiring a token. - Created `TestCloudAuth` class (not skipped) for the `test_token_file_permissions` unit test. - This ensures the security test runs in all environments while keeping integration tests skipped when no token is present. Co-authored-by: refraction-ray <35157286+refraction-ray@users.noreply.github.com> --- tests/test_cloud.py | 497 +++++++++++++++++++++----------------------- 1 file changed, 234 insertions(+), 263 deletions(-) diff --git a/tests/test_cloud.py b/tests/test_cloud.py index 17dae8cd..40d63986 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -17,268 +17,239 @@ skip_cloud = pytest.mark.skipif("TC_CLOUD_TEST" not in os.environ, reason="no token") - -@skip_cloud -def test_get_token(): - print(apis.get_token(provider="Tencent")) - p = apis.get_provider("tencent") - print(p.get_token()) - print(p.get_device("simulator:tc").get_token()) - - -@skip_cloud -def test_list_devices(): - print(apis.list_devices()) - p = apis.get_provider() - print(p.list_devices()) - print(p.list_devices(state="on")) - - @skip_cloud -def test_get_device(): - d1 = apis.get_device(device="tencent::hello") - assert d1.name == "hello" - assert d1.provider.name == "tencent" - d2 = apis.get_device(device="hello") - assert d2.name == "hello" - assert d2.provider.name == "tencent" - p = apis.get_provider() - d3 = p.get_device("tencent::hello") - assert d3.name == "hello" - assert d3.provider.name == "tencent" - d4 = p.get_device("hello") - assert d4.name == "hello" - assert d4.provider.name == "tencent" - - -@skip_cloud -def test_get_device_cache(): - d1 = apis.get_device("local::testing") - d2 = apis.get_device(provider="local", device="testing") - apis.set_provider("local") - d3 = apis.get_device("testing") - assert id(d1) == id(d2) - assert id(d3) == id(d1) - apis.set_provider("tencent") - d4 = apis.get_device("testing") - assert d4.provider.name == "tencent" - assert id(d4) != id(d1) - - -@skip_cloud -def test_list_properties(): - d = apis.get_device(device="simulator:aer") - print(d.list_properties()) - print(apis.list_properties(device="simulator:aer")) - with pytest.raises(ValueError): - apis.list_properties(device="hell") - - -@skip_cloud -def test_submit_task(): - c = tc.Circuit(3) - c.H(0) - c.H(1) - c.H(2) - t = apis.submit_task(device="simulator:tc", circuit=c) - r = t.details() - assert r["state"] in ["pending", "completed"] - print(t.results(blocked=True)) - assert t.get_logical_physical_mapping() == {0: 0, 1: 1, 2: 2} - - -@skip_cloud -def test_resubmit_task(): - c = tc.Circuit(3) - c.H(0) - c.H(1) - t = apis.submit_task(device="simulator:aer", circuit=c) - time.sleep(15) - t1 = apis.resubmit_task(t) - print(t.details()) - print(t1.details(wait=True)) - - -@skip_cloud -def test_get_task(): - apis.set_device("simulator:tcn1") - c = tc.Circuit(2) - c.cx(0, 1) - t = apis.submit_task(circuit=c) - t1 = apis.get_task(t.id_) - assert t1.id_ == t.id_ - t2 = apis.get_device("tencent::simulator:tcn1").get_task(t.id_) - assert t2.id_ == t.id_ - - apis.set_device() - - -@skip_cloud -def test_list_tasks(): - d = apis.get_device(device="simulator:aer") - print(d.list_tasks()) - print(apis.list_tasks(device="simulator:tc")) - - -@skip_cloud -def test_local_list_device(): - dd = apis.list_devices(provider="local") - assert dd[0].name == "testing" - - -@skip_cloud -def test_local_submit_task(): - c = tc.Circuit(2) - c.h(0) - c.cx(0, 1) - - t = apis.submit_task(device="local::testing", circuit=c, shots=2048) - r = t.results(blocked=True) - assert counts.kl_divergence({"00": 0.5, "11": 0.5}, r) < 0.1 - print(t.details()) - print(t.get_device()) - - -@skip_cloud -def test_local_list_tasks(): - print(apis.list_tasks(provider="local")) - - -@skip_cloud -def test_local_batch_submit(): - apis.set_provider("local") - c = tc.Circuit(2) - c.h(1) - c.ry(1, theta=0.8) - - ts = apis.submit_task(device="testing", circuit=[c, c]) - print(ts[0].results()) - - apis.set_device("testing") - ts = apis.submit_task(circuit=[c, c]) - print(ts[1].results()) - print(ts[1].details()) - apis.set_provider("tencent") - - -@skip_cloud -def test_batch_exp_ps(): - pss = [[1, 0], [0, 3]] - c = tc.Circuit(2) - c.h(0) - c.x(1) - np.testing.assert_allclose(wrapper.batch_expectation_ps(c, pss), [1, -1], atol=1e-5) - np.testing.assert_allclose( - wrapper.batch_expectation_ps(c, pss, ws=[1, -0.5]), 1.5, atol=1e-5 - ) - np.testing.assert_allclose( - wrapper.batch_expectation_ps(c, pss, device="simulator:tcn1"), - [1, -1], - atol=1e-1, - ) - np.testing.assert_allclose( - wrapper.batch_expectation_ps(c, pss, device="local::default", with_rem=False), - [1, -1], - atol=1e-1, - ) - - -@skip_cloud -def test_batch_submit_template(): - run = tc.cloud.wrapper.batch_submit_template( - device="simulator:tc", batch_limit=2, prior=10 - ) - cs = [] - for i in range(4): - c = tc.Circuit(4) - c.h(i) - cs.append(c) - - rs = run(cs[:3], prior=1) - assert len(rs) == 3 - rs = run(cs[:4]) - assert len(rs) == 4 - - -@skip_cloud -def test_allz_batch(): - n = 5 - - def makec(inputs, params): - c = tc.Circuit(n) - for i in range(n): - c.rx(i, theta=inputs[i]) - for i in range(n): - c.rz(i, theta=params[0, i]) - for i in range(n - 1): - c.cx(i, i + 1) +class TestCloud: + def test_get_token(self): + print(apis.get_token(provider="Tencent")) + p = apis.get_provider("tencent") + print(p.get_token()) + print(p.get_device("simulator:tc").get_token()) + + def test_list_devices(self): + print(apis.list_devices()) + p = apis.get_provider() + print(p.list_devices()) + print(p.list_devices(state="on")) + + def test_get_device(self): + d1 = apis.get_device(device="tencent::hello") + assert d1.name == "hello" + assert d1.provider.name == "tencent" + d2 = apis.get_device(device="hello") + assert d2.name == "hello" + assert d2.provider.name == "tencent" + p = apis.get_provider() + d3 = p.get_device("tencent::hello") + assert d3.name == "hello" + assert d3.provider.name == "tencent" + d4 = p.get_device("hello") + assert d4.name == "hello" + assert d4.provider.name == "tencent" + + def test_get_device_cache(self): + d1 = apis.get_device("local::testing") + d2 = apis.get_device(provider="local", device="testing") + apis.set_provider("local") + d3 = apis.get_device("testing") + assert id(d1) == id(d2) + assert id(d3) == id(d1) + apis.set_provider("tencent") + d4 = apis.get_device("testing") + assert d4.provider.name == "tencent" + assert id(d4) != id(d1) + + def test_list_properties(self): + d = apis.get_device(device="simulator:aer") + print(d.list_properties()) + print(apis.list_properties(device="simulator:aer")) + with pytest.raises(ValueError): + apis.list_properties(device="hell") + + def test_submit_task(self): + c = tc.Circuit(3) + c.H(0) + c.H(1) + c.H(2) + t = apis.submit_task(device="simulator:tc", circuit=c) + r = t.details() + assert r["state"] in ["pending", "completed"] + print(t.results(blocked=True)) + assert t.get_logical_physical_mapping() == {0: 0, 1: 1, 2: 2} + + def test_resubmit_task(self): + c = tc.Circuit(3) + c.H(0) + c.H(1) + t = apis.submit_task(device="simulator:aer", circuit=c) + time.sleep(15) + t1 = apis.resubmit_task(t) + print(t.details()) + print(t1.details(wait=True)) + + def test_get_task(self): + apis.set_device("simulator:tcn1") + c = tc.Circuit(2) + c.cx(0, 1) + t = apis.submit_task(circuit=c) + t1 = apis.get_task(t.id_) + assert t1.id_ == t.id_ + t2 = apis.get_device("tencent::simulator:tcn1").get_task(t.id_) + assert t2.id_ == t.id_ + + apis.set_device() + + def test_list_tasks(self): + d = apis.get_device(device="simulator:aer") + print(d.list_tasks()) + print(apis.list_tasks(device="simulator:tc")) + + def test_local_list_device(self): + dd = apis.list_devices(provider="local") + assert dd[0].name == "testing" + + def test_local_submit_task(self): + c = tc.Circuit(2) + c.h(0) + c.cx(0, 1) + + t = apis.submit_task(device="local::testing", circuit=c, shots=2048) + r = t.results(blocked=True) + assert counts.kl_divergence({"00": 0.5, "11": 0.5}, r) < 0.1 + print(t.details()) + print(t.get_device()) + + def test_local_list_tasks(self): + print(apis.list_tasks(provider="local")) + + def test_local_batch_submit(self): + apis.set_provider("local") + c = tc.Circuit(2) + c.h(1) + c.ry(1, theta=0.8) + + ts = apis.submit_task(device="testing", circuit=[c, c]) + print(ts[0].results()) + + apis.set_device("testing") + ts = apis.submit_task(circuit=[c, c]) + print(ts[1].results()) + print(ts[1].details()) + apis.set_provider("tencent") + + def test_batch_exp_ps(self): + pss = [[1, 0], [0, 3]] + c = tc.Circuit(2) + c.h(0) + c.x(1) + np.testing.assert_allclose(wrapper.batch_expectation_ps(c, pss), [1, -1], atol=1e-5) + np.testing.assert_allclose( + wrapper.batch_expectation_ps(c, pss, ws=[1, -0.5]), 1.5, atol=1e-5 + ) + np.testing.assert_allclose( + wrapper.batch_expectation_ps(c, pss, device="simulator:tcn1"), + [1, -1], + atol=1e-1, + ) + np.testing.assert_allclose( + wrapper.batch_expectation_ps(c, pss, device="local::default", with_rem=False), + [1, -1], + atol=1e-1, + ) + + def test_batch_submit_template(self): + run = tc.cloud.wrapper.batch_submit_template( + device="simulator:tc", batch_limit=2, prior=10 + ) + cs = [] + for i in range(4): + c = tc.Circuit(4) + c.h(i) + cs.append(c) + + rs = run(cs[:3], prior=1) + assert len(rs) == 3 + rs = run(cs[:4]) + assert len(rs) == 4 + + def test_allz_batch(self): + n = 5 + + def makec(inputs, params): + c = tc.Circuit(n) + for i in range(n): + c.rx(i, theta=inputs[i]) + for i in range(n): + c.rz(i, theta=params[0, i]) + for i in range(n - 1): + c.cx(i, i + 1) + for i in range(n): + c.rx(i, theta=params[1, i]) + return c + + pss = [] for i in range(n): - c.rx(i, theta=params[1, i]) - return c - - pss = [] - for i in range(n): - ps = [0 for _ in range(n)] - ps[i] = 3 - pss.append(ps) - - def exp_val(c, device=None): - rs = tc.cloud.wrapper.batch_expectation_ps(c, pss, device) - return tc.backend.stack(rs) - - def qmlf(inputs, params, device=None): - c = makec(inputs, params) - return exp_val(c, device) - - inputs = tc.array_to_tensor(np.array([0, 1, 0, 1, 0])) - params = np.ones([2, n]) - print(qmlf(inputs, params, device="9gmon")) - print(qmlf(inputs, params)) - - -def test_token_file_permissions(tmp_path): - # Mock os.path.expanduser to return tmp_path - # We patch where it is used in tensorcircuit.cloud.apis - with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): - authpath = tmp_path / ".tc.auth.json" - - # Ensure clean state for saved_token global variable if necessary - # set_token with clear=True clears the global saved_token - apis.set_token(clear=True) - - # Scenario 1: File creation (new file) - # Set a dummy token - apis.set_token(token="dummy_token_1", provider="tencent", cached=True) - - # Verify file exists - assert authpath.exists() - - # Verify permissions (only on POSIX systems where these bits are meaningful) - if os.name == "posix": - st = os.stat(authpath) - # Check that group and others have no permissions (should be 0) - assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 - - # Scenario 2: File update (existing file with insecure permissions) - if os.name == "posix": - # Manually set insecure permissions to simulate an existing insecure file - os.chmod(authpath, 0o666) - st_before = os.stat(authpath) - # Verify it is indeed insecure (readable by others) - assert (st_before.st_mode & stat.S_IROTH) - - # Update token (add another token or update existing) - # This triggers the write logic again - apis.set_token(token="dummy_token_2", provider="local", cached=True) - - # Verify permissions again (should be fixed to 0600) - if os.name == "posix": - st_after = os.stat(authpath) - assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 - - # Verify content updated and readable - with open(authpath, "r") as f: - data = json.load(f) - assert data is not None + ps = [0 for _ in range(n)] + ps[i] = 3 + pss.append(ps) + + def exp_val(c, device=None): + rs = tc.cloud.wrapper.batch_expectation_ps(c, pss, device) + return tc.backend.stack(rs) + + def qmlf(inputs, params, device=None): + c = makec(inputs, params) + return exp_val(c, device) + + inputs = tc.array_to_tensor(np.array([0, 1, 0, 1, 0])) + params = np.ones([2, n]) + print(qmlf(inputs, params, device="9gmon")) + print(qmlf(inputs, params)) + + +class TestCloudAuth: + def test_token_file_permissions(self, tmp_path): + # Mock os.path.expanduser to return tmp_path + # We patch where it is used in tensorcircuit.cloud.apis + with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): + authpath = tmp_path / ".tc.auth.json" + + # Ensure clean state for saved_token global variable if necessary + # set_token with clear=True clears the global saved_token + apis.set_token(clear=True) + + # Scenario 1: File creation (new file) + # Set a dummy token + apis.set_token(token="dummy_token_1", provider="tencent", cached=True) + + # Verify file exists + assert authpath.exists() + + # Verify permissions (only on POSIX systems where these bits are meaningful) + if os.name == "posix": + st = os.stat(authpath) + # Check that group and others have no permissions (should be 0) + assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Scenario 2: File update (existing file with insecure permissions) + if os.name == "posix": + # Manually set insecure permissions to simulate an existing insecure file + os.chmod(authpath, 0o666) + st_before = os.stat(authpath) + # Verify it is indeed insecure (readable by others) + assert (st_before.st_mode & stat.S_IROTH) + + # Update token (add another token or update existing) + # This triggers the write logic again + apis.set_token(token="dummy_token_2", provider="local", cached=True) + + # Verify permissions again (should be fixed to 0600) + if os.name == "posix": + st_after = os.stat(authpath) + assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Verify content updated and readable + with open(authpath, "r") as f: + data = json.load(f) + assert data is not None From 6054413b8eadb99af15052bfe29186adc2f6fdff Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 14:53:25 +0000 Subject: [PATCH 4/6] fix(test): prevent state pollution in test_cloud.py - The `TestCloudAuth` class was modifying the global `saved_token` variable in `tensorcircuit.cloud.apis` without restoring it. - This caused subsequent integration tests (which run when `TC_CLOUD_TEST` is set) to fail because they relied on the original token state. - Added a `finally` block to `test_token_file_permissions` to restore `apis.saved_token` to its original value. Co-authored-by: refraction-ray <35157286+refraction-ray@users.noreply.github.com> --- tests/test_cloud.py | 104 +++++++++++++++++++++++++------------------- 1 file changed, 59 insertions(+), 45 deletions(-) diff --git a/tests/test_cloud.py b/tests/test_cloud.py index 40d63986..5b9f1399 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -208,48 +208,62 @@ def qmlf(inputs, params, device=None): class TestCloudAuth: def test_token_file_permissions(self, tmp_path): - # Mock os.path.expanduser to return tmp_path - # We patch where it is used in tensorcircuit.cloud.apis - with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): - authpath = tmp_path / ".tc.auth.json" - - # Ensure clean state for saved_token global variable if necessary - # set_token with clear=True clears the global saved_token - apis.set_token(clear=True) - - # Scenario 1: File creation (new file) - # Set a dummy token - apis.set_token(token="dummy_token_1", provider="tencent", cached=True) - - # Verify file exists - assert authpath.exists() - - # Verify permissions (only on POSIX systems where these bits are meaningful) - if os.name == "posix": - st = os.stat(authpath) - # Check that group and others have no permissions (should be 0) - assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 - - # Scenario 2: File update (existing file with insecure permissions) - if os.name == "posix": - # Manually set insecure permissions to simulate an existing insecure file - os.chmod(authpath, 0o666) - st_before = os.stat(authpath) - # Verify it is indeed insecure (readable by others) - assert (st_before.st_mode & stat.S_IROTH) - - # Update token (add another token or update existing) - # This triggers the write logic again - apis.set_token(token="dummy_token_2", provider="local", cached=True) - - # Verify permissions again (should be fixed to 0600) - if os.name == "posix": - st_after = os.stat(authpath) - assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 - - # Verify content updated and readable - with open(authpath, "r") as f: - data = json.load(f) - assert data is not None + # Save original saved_token to restore it later + original_saved_token = apis.saved_token.copy() + try: + # Mock os.path.expanduser to return tmp_path + # We patch where it is used in tensorcircuit.cloud.apis + with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): + authpath = tmp_path / ".tc.auth.json" + + # Ensure clean state for saved_token global variable if necessary + # set_token with clear=True clears the global saved_token + apis.set_token(clear=True) + + # Scenario 1: File creation (new file) + # Set a dummy token + apis.set_token(token="dummy_token_1", provider="tencent", cached=True) + + # Verify file exists + assert authpath.exists() + + # Verify permissions (only on POSIX systems where these bits are meaningful) + if os.name == "posix": + st = os.stat(authpath) + # Check that group and others have no permissions (should be 0) + assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Scenario 2: File update (existing file with insecure permissions) + if os.name == "posix": + # Manually set insecure permissions to simulate an existing insecure file + os.chmod(authpath, 0o666) + st_before = os.stat(authpath) + # Verify it is indeed insecure (readable by others) + assert (st_before.st_mode & stat.S_IROTH) + + # Update token (add another token or update existing) + # This triggers the write logic again + apis.set_token(token="dummy_token_2", provider="local", cached=True) + + # Verify permissions again (should be fixed to 0600) + if os.name == "posix": + st_after = os.stat(authpath) + assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | + stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + + # Verify content updated and readable + with open(authpath, "r") as f: + data = json.load(f) + assert data is not None + finally: + # Restore original saved_token + # We must update the global variable in apis module + # We can't assign to it directly from here easily if it's not imported as mutable + # But apis.saved_token IS mutable (it's a dict). + # However, apis.set_token(clear=True) might reassign the variable inside apis.py: + # "saved_token = {}" inside set_token might create a new dict, but global saved_token refers to it. + # Wait, `from . import saved_token` would import a reference, but `import tensorcircuit.cloud.apis as apis` -> `apis.saved_token` accesses the module attribute. + # If set_token does `global saved_token; saved_token = {}`, it rebinds the name in module scope. + # So we can just reassign it back. + apis.saved_token = original_saved_token From 0f03c0b22ade3c2ca693e77473229f75a3dfd140 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 15:03:54 +0000 Subject: [PATCH 5/6] refactor(test): cleanup verbose comments in test_cloud.py - Removed "thinking out loud" comments from the teardown block in `TestCloudAuth.test_token_file_permissions`. Co-authored-by: refraction-ray <35157286+refraction-ray@users.noreply.github.com> --- tests/test_cloud.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/tests/test_cloud.py b/tests/test_cloud.py index 5b9f1399..f2ce4071 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -257,13 +257,4 @@ def test_token_file_permissions(self, tmp_path): data = json.load(f) assert data is not None finally: - # Restore original saved_token - # We must update the global variable in apis module - # We can't assign to it directly from here easily if it's not imported as mutable - # But apis.saved_token IS mutable (it's a dict). - # However, apis.set_token(clear=True) might reassign the variable inside apis.py: - # "saved_token = {}" inside set_token might create a new dict, but global saved_token refers to it. - # Wait, `from . import saved_token` would import a reference, but `import tensorcircuit.cloud.apis as apis` -> `apis.saved_token` accesses the module attribute. - # If set_token does `global saved_token; saved_token = {}`, it rebinds the name in module scope. - # So we can just reassign it back. apis.saved_token = original_saved_token From 792dc733e971a4a17cf8a7b008c1ddf256e47596 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 15:39:40 +0000 Subject: [PATCH 6/6] fix(style): format test_cloud.py to pass check_all.sh - Applied black formatting to `tests/test_cloud.py`. - Fixed pylint import order warnings. Co-authored-by: refraction-ray <35157286+refraction-ray@users.noreply.github.com> --- tests/test_cloud.py | 46 +++++++++++++++++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 10 deletions(-) diff --git a/tests/test_cloud.py b/tests/test_cloud.py index f2ce4071..0b71b3f6 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -1,11 +1,11 @@ import sys import os import time -import pytest -import numpy as np import stat import json from unittest.mock import patch +import pytest +import numpy as np thisfile = os.path.abspath(__file__) modulepath = os.path.dirname(os.path.dirname(thisfile)) @@ -17,6 +17,7 @@ skip_cloud = pytest.mark.skipif("TC_CLOUD_TEST" not in os.environ, reason="no token") + @skip_cloud class TestCloud: def test_get_token(self): @@ -141,7 +142,9 @@ def test_batch_exp_ps(self): c = tc.Circuit(2) c.h(0) c.x(1) - np.testing.assert_allclose(wrapper.batch_expectation_ps(c, pss), [1, -1], atol=1e-5) + np.testing.assert_allclose( + wrapper.batch_expectation_ps(c, pss), [1, -1], atol=1e-5 + ) np.testing.assert_allclose( wrapper.batch_expectation_ps(c, pss, ws=[1, -0.5]), 1.5, atol=1e-5 ) @@ -151,7 +154,9 @@ def test_batch_exp_ps(self): atol=1e-1, ) np.testing.assert_allclose( - wrapper.batch_expectation_ps(c, pss, device="local::default", with_rem=False), + wrapper.batch_expectation_ps( + c, pss, device="local::default", with_rem=False + ), [1, -1], atol=1e-1, ) @@ -213,7 +218,10 @@ def test_token_file_permissions(self, tmp_path): try: # Mock os.path.expanduser to return tmp_path # We patch where it is used in tensorcircuit.cloud.apis - with patch("tensorcircuit.cloud.apis.os.path.expanduser", return_value=str(tmp_path)): + with patch( + "tensorcircuit.cloud.apis.os.path.expanduser", + return_value=str(tmp_path), + ): authpath = tmp_path / ".tc.auth.json" # Ensure clean state for saved_token global variable if necessary @@ -231,8 +239,17 @@ def test_token_file_permissions(self, tmp_path): if os.name == "posix": st = os.stat(authpath) # Check that group and others have no permissions (should be 0) - assert (st.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + assert ( + st.st_mode + & ( + stat.S_IRGRP + | stat.S_IWGRP + | stat.S_IXGRP + | stat.S_IROTH + | stat.S_IWOTH + | stat.S_IXOTH + ) + ) == 0 # Scenario 2: File update (existing file with insecure permissions) if os.name == "posix": @@ -240,7 +257,7 @@ def test_token_file_permissions(self, tmp_path): os.chmod(authpath, 0o666) st_before = os.stat(authpath) # Verify it is indeed insecure (readable by others) - assert (st_before.st_mode & stat.S_IROTH) + assert st_before.st_mode & stat.S_IROTH # Update token (add another token or update existing) # This triggers the write logic again @@ -249,8 +266,17 @@ def test_token_file_permissions(self, tmp_path): # Verify permissions again (should be fixed to 0600) if os.name == "posix": st_after = os.stat(authpath) - assert (st_after.st_mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP | - stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)) == 0 + assert ( + st_after.st_mode + & ( + stat.S_IRGRP + | stat.S_IWGRP + | stat.S_IXGRP + | stat.S_IROTH + | stat.S_IWOTH + | stat.S_IXOTH + ) + ) == 0 # Verify content updated and readable with open(authpath, "r") as f: