#!/usr/bin/env python3
# Copyright 2021 The Pigweed Authors
#
# 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
#
#     https://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.
"""Tests for the thread analyzer."""

import unittest
from pw_thread.thread_analyzer import ThreadInfo, ThreadSnapshotAnalyzer
from pw_thread_protos import thread_pb2
import pw_tokenizer
from pw_tokenizer import tokens


class ThreadInfoTest(unittest.TestCase):
    """Tests that the ThreadInfo class produces expected results."""

    def test_empty_thread(self):
        thread_info = ThreadInfo(thread_pb2.Thread())
        expected = '\n'.join(
            (
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
            )
        )
        self.assertFalse(thread_info.has_stack_size_limit())
        self.assertFalse(thread_info.has_stack_used())
        self.assertEqual(expected, str(thread_info))

    def test_thread_with_cpu_usage(self):
        thread = thread_pb2.Thread()
        thread.cpu_usage_hundredths = 1234
        thread_info = ThreadInfo(thread)

        expected = '\n'.join(
            (
                'Est CPU usage: 12.34%',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
            )
        )
        self.assertFalse(thread_info.has_stack_size_limit())
        self.assertFalse(thread_info.has_stack_used())
        self.assertEqual(expected, str(thread_info))

    def test_thread_with_stack_pointer(self):
        thread = thread_pb2.Thread()
        thread.stack_pointer = 0x5AC6A86C
        thread_info = ThreadInfo(thread)

        expected = '\n'.join(
            (
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x5ac6a86c (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
            )
        )
        self.assertFalse(thread_info.has_stack_size_limit())
        self.assertFalse(thread_info.has_stack_used())
        self.assertEqual(expected, str(thread_info))

    def test_thread_with_stack_usage(self):
        thread = thread_pb2.Thread()
        thread.stack_start_pointer = 0x5AC6B86C
        thread.stack_pointer = 0x5AC6A86C
        thread_info = ThreadInfo(thread)

        expected = '\n'.join(
            (
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x5ac6b86c - 0x5ac6a86c (4096 bytes)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x5ac6b86c - 0x???????? (size unknown)',
            )
        )
        self.assertFalse(thread_info.has_stack_size_limit())
        self.assertTrue(thread_info.has_stack_used())
        self.assertEqual(expected, str(thread_info))

    def test_thread_with_all_stack_info(self):
        thread = thread_pb2.Thread()
        thread.stack_start_pointer = 0x5AC6B86C
        thread.stack_end_pointer = 0x5AC6986C
        thread.stack_pointer = 0x5AC6A86C
        thread_info = ThreadInfo(thread)

        # pylint: disable=line-too-long
        expected = '\n'.join(
            (
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x5ac6b86c - 0x5ac6a86c (4096 bytes, 50.00%)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x5ac6b86c - 0x5ac6986c (8192 bytes)',
            )
        )
        # pylint: enable=line-too-long
        self.assertTrue(thread_info.has_stack_size_limit())
        self.assertTrue(thread_info.has_stack_used())
        self.assertEqual(expected, str(thread_info))


class ThreadSnapshotAnalyzerTest(unittest.TestCase):
    """Tests that the ThreadSnapshotAnalyzer class produces expected results."""

    def test_no_threads(self):
        analyzer = ThreadSnapshotAnalyzer(thread_pb2.SnapshotThreadInfo())
        self.assertEqual('', str(analyzer))

    def test_one_empty_thread(self):
        snapshot = thread_pb2.SnapshotThreadInfo()
        snapshot.threads.append(thread_pb2.Thread())
        expected = '\n'.join(
            (
                'Thread State',
                '  1 thread running.',
                '',
                'Thread (UNKNOWN): [unnamed thread]',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
                '',
            )
        )
        analyzer = ThreadSnapshotAnalyzer(snapshot)
        self.assertEqual(analyzer.active_thread(), None)
        self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected)

    def test_two_threads(self):
        """Ensures multiple threads are printed correctly."""
        snapshot = thread_pb2.SnapshotThreadInfo()

        temp_thread = thread_pb2.Thread()
        temp_thread.name = 'Idle'.encode()
        temp_thread.state = thread_pb2.ThreadState.Enum.READY
        temp_thread.stack_start_pointer = 0x2001AC00
        temp_thread.stack_end_pointer = 0x2001AA00
        temp_thread.stack_pointer = 0x2001AB0C
        temp_thread.stack_pointer_est_peak = 0x2001AA00
        snapshot.threads.append(temp_thread)

        temp_thread = thread_pb2.Thread()
        temp_thread.name = 'Alice'.encode()
        temp_thread.stack_start_pointer = 0x2001B000
        temp_thread.stack_pointer = 0x2001AE20
        temp_thread.state = thread_pb2.ThreadState.Enum.BLOCKED
        snapshot.threads.append(temp_thread)

        # pylint: disable=line-too-long
        expected = '\n'.join(
            (
                'Thread State',
                '  2 threads running.',
                '',
                'Thread (READY): Idle',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x2001ac00 - 0x2001ab0c (244 bytes, 47.66%)',
                '  Est peak usage:  512 bytes, 100.00%',
                '  Stack limits:    0x2001ac00 - 0x2001aa00 (512 bytes)',
                '',
                'Thread (BLOCKED): Alice',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x2001b000 - 0x2001ae20 (480 bytes)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x2001b000 - 0x???????? (size unknown)',
                '',
            )
        )
        # pylint: enable=line-too-long
        analyzer = ThreadSnapshotAnalyzer(snapshot)
        self.assertEqual(analyzer.active_thread(), None)
        self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected)

    def test_interrupts_with_thread(self):
        """Ensures interrupts are properly reported as active."""
        snapshot = thread_pb2.SnapshotThreadInfo()

        temp_thread = thread_pb2.Thread()
        temp_thread.name = 'Idle'.encode()
        temp_thread.state = thread_pb2.ThreadState.Enum.READY
        temp_thread.stack_start_pointer = 0x2001AC00
        temp_thread.stack_end_pointer = 0x2001AA00
        temp_thread.stack_pointer = 0x2001AB0C
        temp_thread.stack_pointer_est_peak = 0x2001AA00
        snapshot.threads.append(temp_thread)

        temp_thread = thread_pb2.Thread()
        temp_thread.name = 'Main/Handler'.encode()
        temp_thread.stack_start_pointer = 0x2001B000
        temp_thread.stack_pointer = 0x2001AE20
        temp_thread.state = thread_pb2.ThreadState.Enum.INTERRUPT_HANDLER
        snapshot.threads.append(temp_thread)

        # pylint: disable=line-too-long
        expected = '\n'.join(
            (
                'Thread State',
                '  2 threads running, Main/Handler active at the time of capture.',
                '                     ~~~~~~~~~~~~',
                '',
                # Ensure the active thread is moved to the top of the list.
                'Thread (INTERRUPT_HANDLER): Main/Handler <-- [ACTIVE]',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x2001b000 - 0x2001ae20 (480 bytes)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x2001b000 - 0x???????? (size unknown)',
                '',
                'Thread (READY): Idle',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x2001ac00 - 0x2001ab0c (244 bytes, 47.66%)',
                '  Est peak usage:  512 bytes, 100.00%',
                '  Stack limits:    0x2001ac00 - 0x2001aa00 (512 bytes)',
                '',
            )
        )
        # pylint: enable=line-too-long
        analyzer = ThreadSnapshotAnalyzer(snapshot)
        self.assertEqual(analyzer.active_thread(), temp_thread)
        self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected)

    def test_active_thread(self):
        """Ensures the 'active' thread is highlighted."""
        snapshot = thread_pb2.SnapshotThreadInfo()

        temp_thread = thread_pb2.Thread()
        temp_thread.name = 'Idle'.encode()
        temp_thread.state = thread_pb2.ThreadState.Enum.READY
        temp_thread.stack_start_pointer = 0x2001AC00
        temp_thread.stack_end_pointer = 0x2001AA00
        temp_thread.stack_pointer = 0x2001AB0C
        temp_thread.stack_pointer_est_peak = 0x2001AC00 + 0x100
        snapshot.threads.append(temp_thread)

        temp_thread = thread_pb2.Thread()
        temp_thread.name = 'Main/Handler'.encode()
        temp_thread.active = True
        temp_thread.stack_start_pointer = 0x2001B000
        temp_thread.stack_pointer = 0x2001AE20
        temp_thread.stack_pointer_est_peak = 0x2001B000 + 0x200
        temp_thread.state = thread_pb2.ThreadState.Enum.INTERRUPT_HANDLER
        snapshot.threads.append(temp_thread)

        # pylint: disable=line-too-long
        expected = '\n'.join(
            (
                'Thread State',
                '  2 threads running, Main/Handler active at the time of capture.',
                '                     ~~~~~~~~~~~~',
                '',
                # Ensure the active thread is moved to the top of the list.
                'Thread (INTERRUPT_HANDLER): Main/Handler <-- [ACTIVE]',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x2001b000 - 0x2001ae20 (480 bytes)',
                '  Est peak usage:  512 bytes',
                '  Stack limits:    0x2001b000 - 0x???????? (size unknown)',
                '',
                'Thread (READY): Idle',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x2001ac00 - 0x2001ab0c (244 bytes, 47.66%)',
                '  Est peak usage:  256 bytes, 50.00%',
                '  Stack limits:    0x2001ac00 - 0x2001aa00 (512 bytes)',
                '',
            )
        )
        # pylint: enable=line-too-long
        analyzer = ThreadSnapshotAnalyzer(snapshot)

        # Ensure the active thread is found.
        self.assertEqual(analyzer.active_thread(), temp_thread)
        self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected)

    def test_tokenized_thread_name(self):
        """Ensures a tokenized thread name is detokenized."""
        snapshot = thread_pb2.SnapshotThreadInfo()
        detokenizer = pw_tokenizer.Detokenizer(
            tokens.Database(
                [
                    tokens.TokenizedStringEntry(
                        0x46BE7497, 'The thread for Kuzco'
                    ),
                ]
            )
        )

        temp_thread = thread_pb2.Thread()
        temp_thread.name = b'\x97\x74\xBE\x46'
        snapshot.threads.append(temp_thread)
        temp_thread.name = b'\x5D\xA8\x66\xAE'
        snapshot.threads.append(temp_thread)

        # pylint: disable=line-too-long
        expected = '\n'.join(
            (
                'Thread State',
                '  2 threads running.',
                '',
                'Thread (UNKNOWN): The thread for Kuzco',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
                '',
                'Thread (UNKNOWN): $Xahmrg==',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
                '',
            )
        )
        # pylint: enable=line-too-long
        analyzer = ThreadSnapshotAnalyzer(snapshot, tokenizer_db=detokenizer)

        # Ensure text dump matches expected contents.
        self.assertEqual(str(analyzer), expected)

    def test_no_db_tokenized_thread_name(self):
        """Ensures a tokenized thread name is detokenized."""
        snapshot = thread_pb2.SnapshotThreadInfo()

        temp_thread = thread_pb2.Thread()
        temp_thread.name = b'\x97\x74\xBE\x46'
        snapshot.threads.append(temp_thread)
        temp_thread.name = b'\x5D\xA8\x66\xAE'
        snapshot.threads.append(temp_thread)

        # pylint: disable=line-too-long
        expected = '\n'.join(
            (
                'Thread State',
                '  2 threads running.',
                '',
                'Thread (UNKNOWN): $l3S+Rg==',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
                '',
                'Thread (UNKNOWN): $Xahmrg==',
                'Est CPU usage: unknown',
                'Stack info',
                '  Current usage:   0x???????? - 0x???????? (size unknown)',
                '  Est peak usage:  size unknown',
                '  Stack limits:    0x???????? - 0x???????? (size unknown)',
                '',
            )
        )
        # pylint: enable=line-too-long
        analyzer = ThreadSnapshotAnalyzer(snapshot)

        # Ensure text dump matches expected contents.
        self.assertEqual(str(analyzer), expected)


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