src/performance/lib/trace_writing/tests/json2fxt_test.py

# Copyright 2026 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unit tests for json2fxt.py."""

import json
import os
import pathlib
import struct
import tempfile
import unittest

from json2fxt import FxtConverter


class Json2FxtTest(unittest.TestCase):
    """Test suite validating JSON to FXT conversion."""

    def test_magic_and_initialization(self) -> None:
        converter = FxtConverter()
        with tempfile.TemporaryDirectory() as tmp_dir:
            json_path = os.path.join(tmp_dir, "test.json")
            fxt_path = os.path.join(tmp_dir, "test.fxt")

            with open(json_path, "w") as f:
                json.dump({"traceEvents": []}, f)

            converter.convert(json_path, fxt_path)

            with open(fxt_path, "rb") as f:
                content = f.read()

            # Word 0: Magic number (0x0016547846040010)
            magic = struct.unpack_from("<Q", content, 0)[0]
            self.assertEqual(magic, 0x0016547846040010)

            # Word 1: Initialization record header
            # Type 1 (Initialization), length 2 words
            init_header = struct.unpack_from("<Q", content, 8)[0]
            self.assertEqual(init_header & 0xF, 1)
            self.assertEqual((init_header >> 4) & 0xFFF, 2)

            # Word 2: Ticks per second (1 GHz)
            ticks_per_sec = struct.unpack_from("<Q", content, 16)[0]
            self.assertEqual(ticks_per_sec, 1000000000)

    def test_golden_cpu_metric_conversion(self) -> None:
        """Verify that converting cpu_metric.json produces identical bytes to cpu_metric.fxt."""
        runtime_deps = (
            pathlib.Path(__file__).resolve().parent.parent.parent
            / "runtime_deps"
        )
        json_path = runtime_deps / "cpu_metric.json"
        golden_fxt_path = runtime_deps / "cpu_metric.fxt"
        if not json_path.exists() or not golden_fxt_path.exists():
            self.skipTest(f"Test data not found in {runtime_deps}")

        converter = FxtConverter()
        with tempfile.TemporaryDirectory() as tmp_dir:
            out_fxt_path = os.path.join(tmp_dir, "cpu_metric.fxt")
            converter.convert(json_path, out_fxt_path)

            with open(golden_fxt_path, "rb") as f_golden, open(
                out_fxt_path, "rb"
            ) as f_out:
                self.assertEqual(f_golden.read(), f_out.read())

    def test_argument_types(self) -> None:
        """Verify encoding of various argument types (int32, int64, float, str, bool, process)."""
        converter = FxtConverter()
        # int32
        arg_i32 = converter._pack_argument("int32_val", 42)
        self.assertGreater(len(arg_i32), 0)
        # int64
        arg_i64 = converter._pack_argument("int64_val", 0x100000000)
        self.assertGreater(len(arg_i64), 0)
        # float
        arg_float = converter._pack_argument("float_val", 3.14)
        self.assertGreater(len(arg_float), 0)
        # str
        arg_str = converter._pack_argument("str_val", "hello")
        self.assertGreater(len(arg_str), 0)
        # bool
        arg_bool = converter._pack_argument("bool_val", True)
        self.assertGreater(len(arg_bool), 0)
        # Verify bool header type is 9
        header = struct.unpack_from("<Q", arg_bool, 0)[0]
        self.assertEqual(header & 0xF, 9)
        # process koid
        arg_proc = converter._pack_argument("process", 1234)
        self.assertGreater(len(arg_proc), 0)
        proc_header = struct.unpack_from("<Q", arg_proc, 0)[0]
        self.assertEqual(proc_header & 0xF, 8)

    def test_string_too_long_raises(self) -> None:
        converter = FxtConverter()
        too_long = "x" * 0x8000
        with self.assertRaises(ValueError):
            converter._make_string_ref(too_long)

    def test_synthetic_events_conversion(self) -> None:
        """Verify that synthetic trace with all event types converts without error."""
        synthetic_trace = {
            "traceEvents": [
                {
                    "ph": "M",
                    "name": "process_name",
                    "pid": 100,
                    "args": {"name": "test_proc"},
                },
                {
                    "ph": "M",
                    "name": "thread_name",
                    "pid": 100,
                    "tid": 200,
                    "args": {"name": "test_thread"},
                },
                {
                    "ph": "X",
                    "cat": "category",
                    "name": "duration_complete",
                    "ts": 1000.0,
                    "dur": 500.0,
                    "pid": 100,
                    "tid": 200,
                    "args": {"count": 1, "flag": True},
                },
                {
                    "ph": "B",
                    "cat": "category",
                    "name": "duration_begin",
                    "ts": 2000.0,
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "E",
                    "cat": "category",
                    "name": "duration_end",
                    "ts": 2500.0,
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "i",
                    "cat": "category",
                    "name": "instant_global",
                    "ts": 3000.0,
                    "s": "g",
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "i",
                    "cat": "category",
                    "name": "instant_thread",
                    "ts": 3100.0,
                    "s": "t",
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "C",
                    "cat": "category",
                    "name": "counter_event",
                    "ts": 4000.0,
                    "pid": 100,
                    "tid": 200,
                    "args": {"series_a": 10},
                },
                {
                    "ph": "b",
                    "cat": "category",
                    "name": "async_event",
                    "ts": 5000.0,
                    "id": 42,
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "e",
                    "cat": "category",
                    "name": "async_event",
                    "ts": 5500.0,
                    "id": 42,
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "s",
                    "cat": "category",
                    "name": "flow_event",
                    "ts": 6000.0,
                    "id": "0xabc",
                    "pid": 100,
                    "tid": 200,
                },
                {
                    "ph": "f",
                    "cat": "category",
                    "name": "flow_event",
                    "ts": 6500.0,
                    "id": "0xabc",
                    "pid": 100,
                    "tid": 200,
                },
            ],
            "systemTraceEvents": {
                "events": [
                    {"ph": "p", "pid": 100, "name": "test_proc"},
                    {"ph": "t", "pid": 100, "tid": 200, "name": "test_thread"},
                    {"ph": "w", "cpu": 0, "ts": 100.0, "tid": 200},
                    {
                        "ph": "k",
                        "cpu": 0,
                        "ts": 105.0,
                        "out": {"tid": 200, "prio": 16, "state": 0},
                        "in": {"tid": 0, "prio": 0},
                    },
                ]
            },
        }

        converter = FxtConverter()
        with tempfile.TemporaryDirectory() as tmp_dir:
            json_path = os.path.join(tmp_dir, "synthetic.json")
            fxt_path = os.path.join(tmp_dir, "synthetic.fxt")

            with open(json_path, "w") as f:
                json.dump(synthetic_trace, f)

            converter.convert(json_path, fxt_path)

            self.assertTrue(os.path.exists(fxt_path))
            self.assertGreater(os.path.getsize(fxt_path), 0)


if __name__ == "__main__":
    unittest.main()