Skip to content

test_affine_quantized_float.py pytest too unittest #2261

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.

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

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open
Changes from 1 commit
Commits
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
Prev Previous commit
Next Next commit
test_bitpacking.py pytest too unittest
  • Loading branch information
ved1beta committed May 25, 2025
commit 59ade08cd9e1ce7cedf2d083aa2c2b516b23ccbf
139 changes: 82 additions & 57 deletions test/dtypes/test_bitpacking.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,78 +3,103 @@
#
# This source code is licensed under the BSD 3-Clause license found in the
# LICENSE file in the root directory of this source tree.
import pytest
import unittest
import torch
from torch.utils._triton import has_triton
from torch.testing._internal import common_utils

from torchao.dtypes.uintx.bitpacking import pack, pack_cpu, unpack, unpack_cpu

bit_widths = (1, 2, 3, 4, 5, 6, 7)
dimensions = (0, -1, 1)


@pytest.fixture(autouse=True)
def run_before_and_after_tests():
yield
torch._dynamo.reset() # reset cache between tests
class TestBitpacking(unittest.TestCase):
def setUp(self):
"""Set up test fixtures before each test method."""
# Set random seeds for reproducibility
torch.manual_seed(0)
if torch.cuda.is_available():
torch.cuda.manual_seed(0)

# Initialize any common test data
self.test_shape = (32, 32, 32)

# Set default device
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# Reset any cached state
torch._dynamo.reset()

def tearDown(self):
"""Clean up after each test method."""
# Reset torch._dynamo cache
torch._dynamo.reset()

# Clear CUDA cache if available
if torch.cuda.is_available():
torch.cuda.empty_cache()

# Reset any other state if needed
torch._dynamo.config.specialize_int = False

@pytest.mark.parametrize("bit_width", bit_widths)
@pytest.mark.parametrize("dim", dimensions)
def test_CPU(bit_width, dim):
test_tensor = torch.randint(
0, 2**bit_width, (32, 32, 32), dtype=torch.uint8, device="cpu"
)
packed = pack_cpu(test_tensor, bit_width, dim=dim)
unpacked = unpack_cpu(packed, bit_width, dim=dim)
assert unpacked.allclose(test_tensor)
@common_utils.parametrize("bit_width", bit_widths)
@common_utils.parametrize("dim", dimensions)
def test_CPU(self, bit_width, dim):
test_tensor = torch.randint(
0, 2**bit_width, (32, 32, 32), dtype=torch.uint8, device="cpu"
)
packed = pack_cpu(test_tensor, bit_width, dim=dim)
unpacked = unpack_cpu(packed, bit_width, dim=dim)
self.assertTrue(unpacked.allclose(test_tensor))

@unittest.skipIf(not torch.cuda.is_available(), "CUDA not available")
@common_utils.parametrize("bit_width", bit_widths)
@common_utils.parametrize("dim", dimensions)
def test_GPU(self, bit_width, dim):
test_tensor = torch.randint(0, 2**bit_width, (32, 32, 32), dtype=torch.uint8).cuda()
packed = pack(test_tensor, bit_width, dim=dim)
unpacked = unpack(packed, bit_width, dim=dim)
self.assertTrue(unpacked.allclose(test_tensor))

@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
@pytest.mark.parametrize("bit_width", bit_widths)
@pytest.mark.parametrize("dim", dimensions)
def test_GPU(bit_width, dim):
test_tensor = torch.randint(0, 2**bit_width, (32, 32, 32), dtype=torch.uint8).cuda()
packed = pack(test_tensor, bit_width, dim=dim)
unpacked = unpack(packed, bit_width, dim=dim)
assert unpacked.allclose(test_tensor)
@unittest.skipIf(not torch.cuda.is_available(), "CUDA not available")
@unittest.skipIf(not has_triton(), "unsupported without triton")
@common_utils.parametrize("bit_width", bit_widths)
@common_utils.parametrize("dim", dimensions)
def test_compile(self, bit_width, dim):
torch._dynamo.config.specialize_int = True
torch.compile(pack, fullgraph=True)
torch.compile(unpack, fullgraph=True)
test_tensor = torch.randint(0, 2**bit_width, (32, 32, 32), dtype=torch.uint8).cuda()
packed = pack(test_tensor, bit_width, dim=dim)
unpacked = unpack(packed, bit_width, dim=dim)
self.assertTrue(unpacked.allclose(test_tensor))

@unittest.skipIf(not torch.cuda.is_available(), "CUDA not available")
def test_pack_example(self):
test_tensor = torch.tensor(
[0x30, 0x29, 0x17, 0x5, 0x20, 0x16, 0x9, 0x22], dtype=torch.uint8
).cuda()
shard_4, shard_2 = pack(test_tensor, 6)
print(shard_4, shard_2)
self.assertTrue(torch.tensor([0, 105, 151, 37], dtype=torch.uint8).cuda().allclose(shard_4))
self.assertTrue(torch.tensor([39, 146], dtype=torch.uint8).cuda().allclose(shard_2))
unpacked = unpack([shard_4, shard_2], 6)
self.assertTrue(unpacked.allclose(test_tensor))

@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
@pytest.mark.skipif(not has_triton(), reason="unsupported without triton")
@pytest.mark.parametrize("bit_width", bit_widths)
@pytest.mark.parametrize("dim", dimensions)
def test_compile(bit_width, dim):
torch._dynamo.config.specialize_int = True
torch.compile(pack, fullgraph=True)
torch.compile(unpack, fullgraph=True)
test_tensor = torch.randint(0, 2**bit_width, (32, 32, 32), dtype=torch.uint8).cuda()
packed = pack(test_tensor, bit_width, dim=dim)
unpacked = unpack(packed, bit_width, dim=dim)
assert unpacked.allclose(test_tensor)
def test_pack_example_CPU(self):
test_tensor = torch.tensor(
[0x30, 0x29, 0x17, 0x5, 0x20, 0x16, 0x9, 0x22], dtype=torch.uint8
)
shard_4, shard_2 = pack(test_tensor, 6)
print(shard_4, shard_2)
self.assertTrue(torch.tensor([0, 105, 151, 37], dtype=torch.uint8).allclose(shard_4))
self.assertTrue(torch.tensor([39, 146], dtype=torch.uint8).allclose(shard_2))
unpacked = unpack([shard_4, shard_2], 6)
self.assertTrue(unpacked.allclose(test_tensor))


# these test cases are for the example pack walk through in the bitpacking.py file
@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
def test_pack_example():
test_tensor = torch.tensor(
[0x30, 0x29, 0x17, 0x5, 0x20, 0x16, 0x9, 0x22], dtype=torch.uint8
).cuda()
shard_4, shard_2 = pack(test_tensor, 6)
print(shard_4, shard_2)
assert torch.tensor([0, 105, 151, 37], dtype=torch.uint8).cuda().allclose(shard_4)
assert torch.tensor([39, 146], dtype=torch.uint8).cuda().allclose(shard_2)
unpacked = unpack([shard_4, shard_2], 6)
assert unpacked.allclose(test_tensor)
common_utils.instantiate_parametrized_tests(TestBitpacking)


def test_pack_example_CPU():
test_tensor = torch.tensor(
[0x30, 0x29, 0x17, 0x5, 0x20, 0x16, 0x9, 0x22], dtype=torch.uint8
)
shard_4, shard_2 = pack(test_tensor, 6)
print(shard_4, shard_2)
assert torch.tensor([0, 105, 151, 37], dtype=torch.uint8).allclose(shard_4)
assert torch.tensor([39, 146], dtype=torch.uint8).allclose(shard_2)
unpacked = unpack([shard_4, shard_2], 6)
assert unpacked.allclose(test_tensor)
if __name__ == "__main__":
unittest.main()