diff --git a/tests/unit/torch/utils/test_distributed.py b/tests/unit/torch/utils/test_distributed.py
new file mode 100644
index 00000000000..8432913440e
--- /dev/null
+++ b/tests/unit/torch/utils/test_distributed.py
@@ -0,0 +1,143 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Single-process contracts only: everything here must hold without an initialized
+# torch.distributed process group.
+
+import os
+
+import pytest
+import torch
+from torch import nn
+
+import modelopt.torch.utils.distributed as dist
+
+
+def test_single_process_defaults():
+    assert not dist.is_initialized()
+    assert dist.backend() is None
+    assert dist.size() == 1
+    assert dist.rank() == 0
+    assert dist.is_master()
+    assert dist.is_last_process()
+
+
+def test_local_rank_from_env(monkeypatch):
+    monkeypatch.setenv("LOCAL_RANK", "3")
+    assert dist.local_rank() == 3
+
+
+def test_local_rank_fallback_warns_and_uses_global_rank(monkeypatch):
+    monkeypatch.delenv("LOCAL_RANK", raising=False)
+    with pytest.warns(UserWarning, match="LOCAL_RANK"):
+        assert dist.local_rank() == 0
+
+
+def test_broadcast_is_identity_in_single_process():
+    obj = {"a": [1, 2], "b": "x"}
+    assert dist.broadcast(obj) is obj
+
+
+def test_allgather_returns_singleton_list():
+    assert dist.allgather(5) == [5]
+    obj = {"k": 1}
+    assert dist.allgather(obj) == [obj]
+
+
+def test_allreduce_sum_and_unsupported_reduction():
+    assert dist.allreduce(5) == 5
+    with pytest.raises(NotImplementedError):
+        dist.allreduce(5, reduction="max")
+
+
+def test_barrier_is_noop():
+    assert dist.barrier() is None
+
+
+def test_master_only_runs_and_returns_result():
+    @dist.master_only
+    def compute():
+        return {"x": 1}
+
+    assert compute() == {"x": 1}
+
+
+def test_serialize_deserialize_round_trip():
+    obj = {"w": torch.arange(4, dtype=torch.float32), "meta": "hi"}
+    tensor = dist._serialize(obj)
+    assert tensor.dtype == torch.uint8
+    restored = dist._deserialize(tensor)
+    assert restored["meta"] == "hi"
+    assert torch.equal(restored["w"], obj["w"])
+    # The size argument slices the buffer back to the payload; note the
+    # deserializer tolerates trailing bytes either way (pickle stops at its
+    # own payload), so this pins correct size handling, not strict trimming.
+    padded = torch.cat([tensor, torch.zeros(8, dtype=torch.uint8)])
+    restored2 = dist._deserialize(padded, size=tensor.numel())
+    assert restored2["meta"] == "hi"
+    assert torch.equal(restored2["w"], obj["w"])
+
+
+@pytest.mark.parametrize("group", [None, -1])
+def test_process_group_uninitialized_defaults(group):
+    pg = dist.DistributedProcessGroup(group)
+    assert not pg.is_initialized()
+    assert pg.rank() == -1
+    assert pg.world_size() == -1
+    assert "initialized: False" in repr(pg)
+
+
+def test_get_dist_syncd_obj_passthrough_when_uninitialized():
+    pg = dist.DistributedProcessGroup(None)
+    # op must NOT be applied for uninitialized groups
+    assert dist.DistributedProcessGroup.get_dist_syncd_obj(7, pg, sum) == 7
+    assert dist.DistributedProcessGroup.get_dist_syncd_obj(7, [pg, pg], max) == 7
+
+
+def test_parallel_state_defaults_and_repr():
+    state = dist.ParallelState()
+    assert not state.data_parallel_group.is_initialized()
+    assert not state.tensor_parallel_group.is_initialized()
+    assert not state.expert_model_parallel_group.is_initialized()
+    assert state.data_parallel_group.world_size() == -1
+    for name in ("data_parallel_group", "tensor_parallel_group", "expert_model_parallel_group"):
+        assert name in repr(state)
+
+
+def test_get_group_returns_none_when_uninitialized():
+    assert dist.get_group([0]) is None
+
+
+def test_is_dtensor_sharded_false_for_plain_model():
+    assert not dist.is_dtensor_sharded(nn.Linear(2, 2))
+
+
+def test_filelock_context_creates_and_removes_lockfile(tmp_path):
+    lock_path = str(tmp_path / "my.lock")
+    with dist.FileLock(lock_path):
+        assert os.path.exists(lock_path)
+    assert not os.path.exists(lock_path)
+
+
+def test_filelock_try_acquire_conflict(tmp_path):
+    lock_path = str(tmp_path / "my.lock")
+    first = dist.FileLock(lock_path)
+    second = dist.FileLock(lock_path)
+    assert first.try_acquire()
+    assert not second.try_acquire()  # already held
+    first.release()
+    assert second.try_acquire()
+    second.release()
+    assert not os.path.exists(lock_path)
diff --git a/tests/unit/torch/utils/test_graph.py b/tests/unit/torch/utils/test_graph.py
new file mode 100644
index 00000000000..4510171e44f
--- /dev/null
+++ b/tests/unit/torch/utils/test_graph.py
@@ -0,0 +1,129 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import torch.nn.functional as F
+from torch import nn
+
+from modelopt.torch.utils.graph import match
+
+
+class LinearRelu(nn.Module):
+    def __init__(self, features=8):
+        super().__init__()
+        self.fc = nn.Linear(features, features)
+        self.act = nn.ReLU()
+
+    def forward(self, x):
+        return self.act(self.fc(x))
+
+
+class LinearSigmoid(nn.Module):
+    def __init__(self, features=8):
+        super().__init__()
+        self.fc = nn.Linear(features, features)
+        self.act = nn.Sigmoid()
+
+    def forward(self, x):
+        return self.act(self.fc(x))
+
+
+class LinearFuncRelu(nn.Module):
+    """Same math as LinearRelu but relu as call_function instead of call_module."""
+
+    def __init__(self, features=8):
+        super().__init__()
+        self.fc = nn.Linear(features, features)
+
+    def forward(self, x):
+        return F.relu(self.fc(x))
+
+
+class LinearMethodRelu(nn.Module):
+    """relu invoked as a tensor method (call_method node)."""
+
+    def __init__(self, features=8):
+        super().__init__()
+        self.fc = nn.Linear(features, features)
+
+    def forward(self, x):
+        return self.fc(x).relu()
+
+
+class JustLinear(nn.Module):
+    def __init__(self, features=8):
+        super().__init__()
+        self.fc = nn.Linear(features, features)
+
+    def forward(self, x):
+        return self.fc(x)
+
+
+class Untraceable(nn.Module):
+    """Data-dependent control flow makes this module untraceable by torch.fx."""
+
+    def forward(self, x):
+        if x.sum() > 0:
+            return x
+        return -x
+
+
+def test_match_same_structure():
+    assert match(LinearRelu(), [LinearRelu()])
+
+
+def test_match_compares_module_types_not_instances():
+    # different layer sizes must still match since submodules are compared by type
+    assert match(LinearRelu(features=4), [LinearRelu(features=16)])
+
+
+def test_no_match_different_activation():
+    assert not match(LinearRelu(), [LinearSigmoid()])
+
+
+def test_no_match_different_node_count():
+    assert not match(JustLinear(), [LinearRelu()])
+    assert not match(LinearRelu(), [JustLinear()])
+
+
+def test_empty_patterns_never_match():
+    assert not match(LinearRelu(), [])
+
+
+def test_match_any_of_multiple_patterns():
+    assert match(LinearRelu(), [LinearSigmoid(), JustLinear(), LinearRelu()])
+
+
+def test_call_function_nodes_match():
+    assert match(LinearFuncRelu(), [LinearFuncRelu()])
+
+
+def test_call_module_does_not_match_call_function():
+    # nn.ReLU (call_module) vs F.relu (call_function) are different graph ops
+    assert not match(LinearRelu(), [LinearFuncRelu()])
+
+
+def test_call_method_nodes_match():
+    assert match(LinearMethodRelu(), [LinearMethodRelu()])
+    assert not match(LinearMethodRelu(), [LinearFuncRelu()])
+
+
+def test_untraceable_module_returns_false():
+    assert not match(Untraceable(), [LinearRelu()])
+
+
+def test_sequential_matches_equivalent_custom_module():
+    seq = nn.Sequential(nn.Linear(8, 8), nn.ReLU())
+    assert match(seq, [LinearRelu()])
+    assert match(LinearRelu(), [seq])
diff --git a/tests/unit/torch/utils/test_import_utils.py b/tests/unit/torch/utils/test_import_utils.py
new file mode 100644
index 00000000000..0fd51aae274
--- /dev/null
+++ b/tests/unit/torch/utils/test_import_utils.py
@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import warnings
+from contextlib import contextmanager
+
+import pytest
+
+from modelopt.torch.utils.import_utils import import_plugin
+
+
+@contextmanager
+def no_warnings():
+    """Fail the test if any warning is emitted inside the block."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        yield
+
+
+def _import_missing_module(tracker=None, **kwargs):
+    """Run import_plugin around an import that always fails with ModuleNotFoundError."""
+    with import_plugin("foo", **kwargs):
+        __import__("definitely_not_a_real_module_xyz")
+        if tracker is not None:
+            tracker.append("ran past import")
+
+
+def _raise_runtime_error(**kwargs):
+    with import_plugin("myplugin", **kwargs):
+        raise RuntimeError("boom")
+
+
+def test_no_error_and_no_success_msg_is_silent():
+    with no_warnings(), import_plugin("foo"):
+        pass
+
+
+def test_success_msg_emitted_on_clean_import():
+    with (
+        pytest.warns(UserWarning, match="foo plugin loaded"),
+        import_plugin("foo", success_msg="foo plugin loaded"),
+    ):
+        pass
+
+
+def test_module_not_found_suppressed_with_message():
+    tracker = []
+    with pytest.warns(UserWarning, match="foo is not installed"):
+        _import_missing_module(tracker=tracker, msg_if_missing="foo is not installed")
+    assert not tracker  # code after the failing import must not run
+
+
+def test_module_not_found_without_message_is_silent():
+    with no_warnings():
+        _import_missing_module()
+
+
+def test_module_not_found_verbose_false_is_silent():
+    with no_warnings():
+        _import_missing_module(msg_if_missing="should not appear", verbose=False)
+
+
+def test_generic_exception_suppressed_with_warning():
+    with pytest.warns(UserWarning, match="Failed to import modelopt myplugin plugin") as record:
+        _raise_runtime_error()
+    assert "RuntimeError('boom')" in str(record[0].message)
+
+
+def test_generic_exception_verbose_false_is_silent():
+    with no_warnings():
+        _raise_runtime_error(verbose=False)
+
+
+def test_success_msg_not_emitted_when_import_fails():
+    with no_warnings():
+        _import_missing_module(success_msg="should not appear", verbose=True)
diff --git a/tests/unit/torch/utils/test_list.py b/tests/unit/torch/utils/test_list.py
new file mode 100644
index 00000000000..19c8735925c
--- /dev/null
+++ b/tests/unit/torch/utils/test_list.py
@@ -0,0 +1,83 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import math
+
+import pytest
+
+from modelopt.torch.utils.list import list_closest_to_median, stats, val2list, val2tuple
+
+
+def test_list_closest_to_median_odd_length():
+    assert list_closest_to_median([1, 2, 3, 4, 5]) == 3
+    assert list_closest_to_median([10, 1, 7]) == 7  # median of [1, 7, 10] is 7
+
+
+def test_list_closest_to_median_even_length_returns_first_closest():
+    # median of [1, 2, 3, 4] is 2.5; 2 and 3 are equally close, first index wins
+    assert list_closest_to_median([1, 2, 3, 4]) == 2
+
+
+def test_list_closest_to_median_singleton_and_floats():
+    assert list_closest_to_median([5]) == 5
+    assert list_closest_to_median([0.1, 0.9, 0.5]) == 0.5
+
+
+def test_val2list_scalar_repeated():
+    assert val2list(3) == [3]
+    assert val2list(3, repeat_time=4) == [3, 3, 3, 3]
+    assert val2list(None, repeat_time=2) == [None, None]
+
+
+def test_val2list_sequences_passed_through():
+    # repeat_time is ignored for list/tuple inputs
+    assert val2list([1, 2], repeat_time=5) == [1, 2]
+    assert val2list((1, 2)) == [1, 2]
+
+
+def test_val2tuple_scalar_expansion():
+    assert val2tuple(3) == (3,)
+    assert val2tuple(3, min_len=3) == (3, 3, 3)
+
+
+def test_val2tuple_repeats_element_at_idx_repeat():
+    # default idx_repeat=-1 repeats the last element (inserted before it)
+    assert val2tuple([1, 2], min_len=4) == (1, 2, 2, 2)
+    # idx_repeat=0 repeats the first element
+    assert val2tuple([1, 2], min_len=4, idx_repeat=0) == (1, 1, 1, 2)
+
+
+def test_val2tuple_no_padding_needed():
+    assert val2tuple([1, 2, 3], min_len=2) == (1, 2, 3)
+    assert val2tuple([], min_len=3) == ()  # empty input stays empty
+
+
+def test_stats_exact_values():
+    result = stats([1.0, 2.0, 3.0, 4.0])
+    assert set(result.keys()) == {"min", "max", "avg", "std"}
+    assert result["min"] == 1.0
+    assert result["max"] == 4.0
+    assert result["avg"] == 2.5
+    assert result["std"] == pytest.approx(math.sqrt(1.25))
+
+
+def test_stats_single_element():
+    result = stats([2.0])
+    assert result["min"] == result["max"] == result["avg"] == 2.0
+    assert result["std"] == 0.0
+
+
+def test_stats_empty_returns_empty_dict():
+    assert stats([]) == {}
diff --git a/tests/unit/torch/utils/test_logging.py b/tests/unit/torch/utils/test_logging.py
new file mode 100644
index 00000000000..7983efe334d
--- /dev/null
+++ b/tests/unit/torch/utils/test_logging.py
@@ -0,0 +1,208 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+import sys
+import warnings
+
+import pytest
+import tqdm
+
+from modelopt.torch.utils.logging import (
+    DeprecatedError,
+    atomic_print,
+    capture_io,
+    no_stdout,
+    num2hrb,
+    print_args,
+    print_rank_0,
+    silence_matched_warnings,
+    warn_rank_0,
+)
+
+
+@pytest.mark.parametrize(
+    ("num", "expected"),
+    [
+        (0, "0.00"),
+        (999, "999.00"),
+        (1000, "1.00K"),
+        (1500, "1.50K"),
+        (-1500, "-1.50K"),
+        (1e6, "1.00M"),
+        (2.5e9, "2.50B"),
+        (1e12, "1.00T"),
+        (1e15, "1.00P"),
+        (1e18, "1.00E"),
+        (1e21, "1000.00E"),  # units exhausted, number keeps growing
+    ],
+)
+def test_num2hrb_exact(num, expected):
+    assert num2hrb(num) == expected
+
+
+def test_num2hrb_suffix():
+    assert num2hrb(1000, suffix="B") == "1.00KB"
+    assert num2hrb(1, suffix="s") == "1.00s"
+
+
+def test_print_rank_0_prints_in_single_process(capsys):
+    print_rank_0("hello", "world")
+    assert capsys.readouterr().out == "hello world\n"
+
+
+def test_print_rank_0_forwards_print_kwargs(capsys):
+    print_rank_0("a", "b", sep="-", end="!")
+    assert capsys.readouterr().out == "a-b!"
+
+
+def test_print_args_sorted_with_header_and_footer(capsys):
+    print_args(argparse.Namespace(beta=2, alpha="x"), title="Test")
+    lines = capsys.readouterr().out.strip().splitlines()
+    header = "=" * 20 + " Test " + "=" * 20
+    assert lines[0] == header
+    assert lines[1] == f"{'alpha':<35} x"  # keys sorted alphabetically
+    assert lines[2] == f"{'beta':<35} 2"
+    assert lines[3] == "=" * len(header)
+
+
+def test_warn_rank_0_emits_user_warning():
+    with pytest.warns(UserWarning, match="something happened"):
+        warn_rank_0("something happened")
+
+
+def test_warn_rank_0_no_ansi_codes_when_stderr_not_a_tty(monkeypatch):
+    # Pin the non-tty branch explicitly so the test also holds under pytest -s
+    # in a real terminal.
+    monkeypatch.setattr(sys.stderr, "isatty", lambda: False)
+    with pytest.warns(UserWarning) as record:
+        warn_rank_0("plain message")
+    assert str(record[0].message) == "plain message"
+
+
+def test_warn_rank_0_stacklevel_points_at_caller():
+    with pytest.warns(UserWarning) as record:
+        warn_rank_0("stack check")
+    assert record[0].filename == __file__
+
+
+def test_no_stdout_silences_prints(capsys):
+    print("before")
+    with no_stdout():
+        print("hidden")
+    print("after")
+    assert capsys.readouterr().out == "before\nafter\n"
+
+
+def test_no_stdout_disables_tqdm_and_restores_it():
+    with no_stdout():
+        bar = tqdm.tqdm(range(2))
+        assert bar.disable
+        bar.close()
+    bar2 = tqdm.tqdm(range(2))
+    assert not bar2.disable  # original tqdm.__init__ restored on exit
+    bar2.close()
+
+
+def test_capture_io_captures_stdout_and_stderr():
+    with capture_io() as buf:
+        print("to stdout")
+        print("to stderr", file=sys.stderr)
+    assert "to stdout" in buf.getvalue()
+    assert "to stderr" in buf.getvalue()
+
+
+def test_capture_io_can_leave_stderr_alone(capsys):
+    with capture_io(capture_stderr=False) as buf:
+        print("out")
+        print("err", file=sys.stderr)
+    assert buf.getvalue() == "out\n"
+    assert "err" in capsys.readouterr().err
+
+
+def test_atomic_print_returns_result_and_prints_once(capsys):
+    @atomic_print
+    def announce():
+        print("part1")
+        print("part2")
+        return 42
+
+    assert announce() == 42
+    assert capsys.readouterr().out == "part1\npart2\n"
+
+
+def test_atomic_print_nested_does_not_deadlock(capsys):
+    @atomic_print
+    def inner():
+        print("inner")
+
+    @atomic_print
+    def outer():
+        print("outer")
+        inner()
+
+    outer()  # reentrant lock allows nesting
+    out = capsys.readouterr().out
+    assert "inner" in out
+    assert "outer" in out
+
+
+def test_atomic_print_restores_stdout_on_exception(capsys):
+    @atomic_print
+    def broken():
+        print("lost output")
+        raise ValueError("boom")
+
+    original_stdout = sys.stdout
+    with pytest.raises(ValueError, match="boom"):
+        broken()
+    assert sys.stdout is original_stdout
+
+
+def test_silence_matched_warnings_filters_by_pattern():
+    seen = []
+
+    def recorder(message, category, filename, lineno, file=None, line=None):
+        seen.append(str(message))
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("always")
+        warnings.showwarning = recorder  # restored by catch_warnings on exit
+        with silence_matched_warnings("skip"):
+            warnings.warn("please skip me")
+            warnings.warn("keep me")
+    assert seen == ["keep me"]
+
+
+def test_silence_matched_warnings_restores_showwarning():
+    original = warnings.showwarning
+    with silence_matched_warnings("pattern"):
+        assert warnings.showwarning is not original
+    assert warnings.showwarning is original
+
+
+@pytest.mark.parametrize("pattern", [None, 123])
+def test_silence_matched_warnings_invalid_pattern_is_noop(pattern):
+    # None or an un-compilable pattern leaves warnings.showwarning untouched
+    original = warnings.showwarning
+    with silence_matched_warnings(pattern):
+        assert warnings.showwarning is original
+    assert warnings.showwarning is original
+
+
+def test_deprecated_error_is_not_implemented_error():
+    assert issubclass(DeprecatedError, NotImplementedError)
+    with pytest.raises(NotImplementedError, match="gone"):
+        raise DeprecatedError("gone")
diff --git a/tests/unit/torch/utils/test_random.py b/tests/unit/torch/utils/test_random.py
new file mode 100644
index 00000000000..4c3e40c28cc
--- /dev/null
+++ b/tests/unit/torch/utils/test_random.py
@@ -0,0 +1,149 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import random as _stdlib_random
+
+import pytest
+
+from modelopt.torch.utils import random as mo_random
+from modelopt.torch.utils.random import _deterministic_seed, _get_generator, _set_deterministic_seed
+
+
+def _clear_generator_state():
+    for attr in ("generator", "is_manual", "is_synced"):
+        if hasattr(_get_generator, attr):
+            delattr(_get_generator, attr)
+
+
+@pytest.fixture(autouse=True)
+def _fresh_generator_state():
+    """Ensure each test starts and ends without a cached module-level generator."""
+    _clear_generator_state()
+    yield
+    _clear_generator_state()
+
+
+def test_manual_seed_matches_stdlib_random():
+    """A manual seed must produce exactly the stdlib ``random.Random(seed)`` stream."""
+    _set_deterministic_seed(1234)
+    expected = _stdlib_random.Random(1234)
+    assert [mo_random.random() for _ in range(5)] == [expected.random() for _ in range(5)]
+
+
+def test_default_seed_is_one():
+    _set_deterministic_seed()
+    assert mo_random.random() == _stdlib_random.Random(1).random()
+
+
+def test_reseeding_reproduces_sequence():
+    _set_deterministic_seed(77)
+    first_run = [mo_random.random() for _ in range(4)]
+    _set_deterministic_seed(77)
+    assert [mo_random.random() for _ in range(4)] == first_run
+
+
+def test_different_seeds_produce_different_sequences():
+    _set_deterministic_seed(1)
+    seq_a = [mo_random.random() for _ in range(4)]
+    _set_deterministic_seed(2)
+    assert [mo_random.random() for _ in range(4)] != seq_a
+
+
+def test_independent_of_global_random_state():
+    """Reseeding the global ``random`` module must not affect the modelopt generator."""
+    _set_deterministic_seed(3)
+    expected = _stdlib_random.Random(3)
+    assert mo_random.random() == expected.random()
+    _stdlib_random.seed(999)  # must have no effect on the modelopt stream
+    assert mo_random.random() == expected.random()
+
+
+def test_random_in_unit_interval_with_auto_seed():
+    # no manual seed -> generator is auto-seeded lazily
+    for _ in range(100):
+        assert 0.0 <= mo_random.random() < 1.0
+
+
+def test_auto_seeded_generator_is_cached():
+    assert _get_generator() is _get_generator()
+
+
+def test_manual_seed_replaces_existing_generator():
+    generator_before = _get_generator()  # auto-seeded
+    _set_deterministic_seed(5)
+    assert _get_generator() is not generator_before
+    assert mo_random.random() == _stdlib_random.Random(5).random()
+
+
+def test_choice_deterministic():
+    seq = list(range(100))
+    _set_deterministic_seed(7)
+    assert mo_random.choice(seq) == _stdlib_random.Random(7).choice(seq)
+
+
+def test_sample_deterministic():
+    _set_deterministic_seed(11)
+    assert mo_random.sample(range(20), 5) == _stdlib_random.Random(11).sample(range(20), 5)
+
+
+def test_shuffle_deterministic_and_in_place():
+    _set_deterministic_seed(42)
+    data = list(range(10))
+    assert mo_random.shuffle(data) is None  # in-place, like random.shuffle
+    expected = list(range(10))
+    _stdlib_random.Random(42).shuffle(expected)
+    assert data == expected
+
+
+def test_centroid_exact_values():
+    # scalars: median of [1, 2, 3, 4, 5] is 3
+    assert mo_random.centroid([1, 2, 3, 4, 5]) == 3
+    # tuples are reduced via torch.prod: [6, 1, 16] -> median 6 -> (2, 3)
+    assert mo_random.centroid([(2, 3), (1, 1), (4, 4)]) == (2, 3)
+
+
+def test_centroid_even_length_returns_first_closest():
+    # median of [1, 2, 3, 4] is 2.5; 2 and 3 are equally close, first match wins
+    assert mo_random.centroid([1, 2, 3, 4]) == 2
+
+
+def test_original_returns_none():
+    assert mo_random.original([1, 2, 3]) is None
+    assert mo_random.original([]) is None
+
+
+def test_deterministic_seed_context_uses_seed_1024_and_restores_state():
+    # NOTE: `_deterministic_seed` does not wrap its body in try/finally, so the previous
+    # generator is NOT restored if the body raises. Only the happy path is tested here.
+    _set_deterministic_seed(99)
+    outer = _stdlib_random.Random(99)
+    inner = _stdlib_random.Random(1024)
+
+    assert mo_random.random() == outer.random()
+    with _deterministic_seed():
+        assert mo_random.random() == inner.random()
+        assert mo_random.random() == inner.random()
+    # the outer stream continues exactly where it left off
+    assert mo_random.random() == outer.random()
+
+
+def test_is_manual_flag_after_seeding():
+    # NOTE: `_get_generator` computes `is_manual = seed is not None` *after* reassigning
+    # `seed = dist.broadcast(seed or _random.getrandbits(64))`, so `is_manual` is True even
+    # for auto-generated seeds. In a distributed run, an auto-seeded generator created before
+    # process-group initialization would therefore never be re-synced. This test documents
+    # the (correct) manual-seed half of that contract only.
+    _set_deterministic_seed(13)
+    assert _get_generator.is_manual is True
diff --git a/tests/unit/torch/utils/test_robust_json.py b/tests/unit/torch/utils/test_robust_json.py
new file mode 100644
index 00000000000..0cb04b209c2
--- /dev/null
+++ b/tests/unit/torch/utils/test_robust_json.py
@@ -0,0 +1,143 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+import dataclasses
+import datetime
+import json
+from enum import Enum
+from pathlib import Path
+
+import torch
+from omegaconf import OmegaConf
+
+from modelopt.torch.utils.robust_json import json_dump, json_dumps, json_load
+
+
+@dataclasses.dataclass
+class _Point:
+    x: int
+    y: int
+
+
+@dataclasses.dataclass
+class _Config:
+    name: str
+    point: _Point
+    out_dir: Path
+
+
+class _Color(Enum):
+    RED = "red"
+    GREEN = "green"
+
+
+class _Widget:
+    """Arbitrary object without any special JSON support."""
+
+    def render(self):
+        """Regular method used for the bound-method encoding test."""
+
+
+def _dumps_loads(obj):
+    return json.loads(json_dumps(obj))
+
+
+def test_plain_json_types_round_trip():
+    obj = {"a": 1, "b": 2.5, "c": "str", "d": None, "e": True, "f": [1, "x", False]}
+    assert _dumps_loads(obj) == obj
+
+
+def test_output_is_indented_by_two_spaces():
+    assert json_dumps({"a": 1}) == '{\n  "a": 1\n}'
+
+
+def test_dataclass_encoded_as_dict_recursively():
+    cfg = _Config(name="run", point=_Point(x=1, y=2), out_dir=Path("out") / "dir")
+    assert _dumps_loads(cfg) == {
+        "name": "run",
+        "point": {"x": 1, "y": 2},
+        "out_dir": str(Path("out") / "dir"),
+    }
+
+
+def test_path_encoded_as_string():
+    path = Path("some") / "nested" / "file.txt"
+    assert _dumps_loads({"p": path}) == {"p": str(path)}
+
+
+def test_enum_encoded_as_name_not_value():
+    assert _dumps_loads({"color": _Color.RED}) == {"color": "RED"}
+
+
+def test_argparse_namespace_encoded_as_dict():
+    ns = argparse.Namespace(lr=0.1, name="exp")
+    assert _dumps_loads(ns) == {"lr": 0.1, "name": "exp"}
+
+
+def test_torch_dtype_encoded_as_string():
+    assert _dumps_loads({"dtype": torch.float32}) == {"dtype": "torch.float32"}
+    assert _dumps_loads({"dtype": torch.bfloat16}) == {"dtype": "torch.bfloat16"}
+
+
+def test_omegaconf_containers_resolved():
+    dict_cfg = OmegaConf.create({"a": 1, "b": {"c": [1, 2]}})
+    list_cfg = OmegaConf.create([1, "x"])
+    assert _dumps_loads(dict_cfg) == {"a": 1, "b": {"c": [1, 2]}}
+    assert _dumps_loads(list_cfg) == [1, "x"]
+
+
+def test_omegaconf_interpolation_resolved():
+    cfg = OmegaConf.create({"a": 5, "b": "${a}"})
+    assert _dumps_loads(cfg) == {"a": 5, "b": 5}
+
+
+def test_function_encoded_as_qualified_name():
+    assert _dumps_loads({"fn": json.dumps}) == {"fn": "json.dumps"}
+
+
+def test_method_encoded_as_qualified_name():
+    method = _Widget().render  # bound method
+    expected = f"{method.__module__}.{method.__qualname__}"
+    assert _dumps_loads({"m": method}) == {"m": expected}
+
+
+def test_class_encoded_as_qualified_name():
+    expected = f"{Path.__module__}.{Path.__qualname__}"
+    assert _dumps_loads({"cls": Path}) == {"cls": expected}
+
+
+def test_timedelta_encoded_as_string():
+    td = datetime.timedelta(hours=1, minutes=2, seconds=3)
+    assert _dumps_loads({"td": td}) == {"td": "1:02:03"}
+
+
+def test_arbitrary_object_falls_back_to_class_path():
+    expected = f"{_Widget.__module__}.{_Widget.__qualname__}"
+    assert _dumps_loads({"obj": _Widget()}) == {"obj": expected}
+
+
+def test_json_dump_creates_parent_dirs_and_json_load_round_trips(tmp_path):
+    path = tmp_path / "a" / "b" / "result.json"  # parents do not exist yet
+    obj = {"name": "test", "values": [1, 2, 3], "nested": {"ok": True}}
+    json_dump(obj, path)
+    assert path.is_file()
+    assert json_load(path) == obj
+
+
+def test_json_dump_accepts_string_path(tmp_path):
+    path = str(tmp_path / "out.json")
+    json_dump({"k": 1}, path)
+    assert json_load(path) == {"k": 1}