Published March 15, 2017 © MIT

PHPUnicorn - Visualizing PHPUnit Tests

Watch PHPUnit tests run in real-time on a Unicorn pHAT (through a Rasperry Pi Zero W).

BeginnerFull instructions provided1 hour4,005

Things used in this project

Hardware components

Raspberry Pi Zero Wireless

2A Micro-USB Power Adapter

Pimoroni Unicorn pHAT

Software apps and online services

PHPUnit

Hand tools and fabrication machines

Soldering iron (generic)

Story

Goal

I wanted to create a desktop gadget to visualize the progress of unit tests run via PHPUnit.

I've named this project PHPUnicorn (by combining "PHPUnit" with "Unicorn pHAT").

Hardware & Assembly

The hardware requirements for this project:

Raspberry Pi Zero Wireless (W)

MicroSD card flashed with Raspbian Lite

2 amp micro-USB power adapter

Unicorn pHAT by Pimoroni

A 2x20 header

A soldering iron

Put the header between the Pi and Unicorn pHAT and solder it into place.

Load Raspbian onto the microSD card, configure networking, enable SSH, and install Python 3. The unicorn-hat Python library will also need to be installed.

Extending PHPUnit

PHPUnit allows you to easily add listeners via the phpunit.xml file:

<?xml version="1.0" encoding="UTF-8"?>
<phpunit bootstrap="vendor/autoload.php">
    <!-- Existing configuration here --->
    <listeners>
        <listener class="ColinODell\PHPUnicorn\PHPUnicornListener">
            <arguments>
                <string>192.168.80.93</string>
                <string>5005</string>
            </arguments>
        </listener>
    </listeners>
</phpunit>

Each listener is notified when tests begin, when they finish, and what the results are. Here's the code I wrote to do just that:

<?php
namespace ColinODell\PHPUnicorn; 
use Exception; 
use PHPUnit_Framework_AssertionFailedError; 
use PHPUnit_Framework_Test; 
use PHPUnit_Framework_TestSuite; 
use PHPUnit_Framework_Warning; 
class PHPUnicornListener extends \PHPUnit_Framework_BaseTestListener 
{ 
   const NO_RESULT = 'N'; 
   const ERROR = 'E'; 
   const FAILURE = 'F'; 
   const INCOMPLETE = 'I'; 
   const RISKY = 'R'; 
   const SKIPPED = 'S'; 
   const PASSED = 'P'; 
   const WARNING = 'W'; 
   const TOTAL = 'T'; 
   const COMPLETED = 'C';
   private $currentTestPassed = false; 
   private $counts = []; 
   /** 
    * @var string 
    */ 
   private $host; 
   /** 
    * @var int 
    */ 
   private $port; 
   /** 
    * @var resource 
    */ 
   private $socket; 
   /** 
    * @param string $host 
    * @param int    $port 
    */ 
   public function __construct($host, $port) 
   { 
       $this->host = $host; 
       $this->port = $port; 
       $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); 
       $this->resetCounts(); 
       $this->broadcast(); 
   } 
   /** 
    * Ensure socket is closed 
    */ 
   public function __destruct() 
   { 
       socket_close($this->socket); 
       $this->socket = null; 
   } 
   /** 
    * {@inheritdoc} 
    */ 
   public function addError(PHPUnit_Framework_Test $test, Exception $e, $time)
   { 
       $this->counts[self::ERROR]++; 
       $this->currentTestPassed = false; 
   } 
   /** 
    * {@inheritdoc} 
    */ 
   public function addFailure(PHPUnit_Framework_Test $test, PHPUnit_Framework_AssertionFailedError $e, $time) { 
       $this->counts[self::FAILURE]++; 
       $this->currentTestPassed = false; 
   } 
   /** 
    * {@inheritdoc} 
    */ 
   public function addWarning(PHPUnit_Framework_Test $test, PHPUnit_Framework_Warning $e, $time) 
   { 
       $this->counts[self::WARNING]++; 
       $this->currentTestPassed = false; 
   } 
   /** 
    * {@inheritdoc} 
    */ 
   public function addIncompleteTest(PHPUnit_Framework_Test $test, Exception $e, $time) { 
       $this->counts[self::INCOMPLETE]++; 
       $this->currentTestPassed = false; 
   } 
   /** 
    * {@inheritdoc} 
    */ 
   public function addRiskyTest(PHPUnit_Framework_Test $test, Exception $e, $time) { 
       $this->counts[self::RISKY]++; 
       $this->currentTestPassed = false; 
   } 
   /** 
    * {@inheritdoc} 
    */ 
   public function addSkippedTest(PHPUnit_Framework_Test $test, Exception $e, $time) 
   { 
       $this->counts[self::SKIPPED]++; 
       $this->currentTestPassed = false; 
   } 
   /** 
    * A test suite has started. 
    * 
    * {@inheritdoc} 
    */ 
   public function startTestSuite(PHPUnit_Framework_TestSuite $suite) 
   { 
       // Test suites can contain child test suites.  This function is always 
       // called with the top-most parent, so use its count method to determine 
       // how many tests there are (it'll count all sub-children recursively). 
       if ($this->counts[self::TOTAL] == 0) { 
           $this->counts[self::TOTAL] = $suite->count(); 
       } 
   } 
   /** 
    * A test suite ended. 
    * 
    * {@inheritdoc} 
    */ 
   public function endTestSuite(PHPUnit_Framework_TestSuite $suite) 
   { 
       // Send the results over to the Pi 
       $this->broadcast(); 
   } 
   /** 
    * A single test started. 
    * 
    * {@inheritdoc} 
    */ 
   public function startTest(PHPUnit_Framework_Test $test) 
   { 
       // There's no method like addPassed(), so we'll assume the test passes 
       // unless one of the other add___() methods are called. 
       $this->currentTestPassed = true; 
   } 
   /** 
    * A test ended. 
    * 
    * {@inheritdoc} 
    */ 
   public function endTest(PHPUnit_Framework_Test $test, $time) 
   { 
       if ($this->currentTestPassed) { 
           $this->counts[self::PASSED]++; 
       } 
       $this->counts[self::COMPLETED]++; 
       $this->broadcast(); 
   } 
   private function resetCounts() 
   { 
       $this->counts = [ 
           self::NO_RESULT => 0, 
           self::ERROR => 0, 
           self::FAILURE => 0, 
           self::WARNING => 0, 
           self::INCOMPLETE => 0, 
           self::RISKY => 0, 
           self::SKIPPED => 0, 
           self::PASSED => 0, 
           self::TOTAL => 0, 
           self::COMPLETED => 0, 
       ]; 
   } 
   private function broadcast() 
   { 
       if ($this->counts[self::TOTAL] == 0) { 
           // Ask the Pi to clear the screen when we first start 
           $message = 'reset'; 
       } else { 
           $message = json_encode($this->counts); 
       } 
       socket_sendto($this->socket, $message, strlen($message), 0, $this->host, $this->port); 
   } 
}

The code is fairly straight-forward - as each test result comes in, we increment the corresponding result in $this->counts and broadcast the latest counts to the Raspberry Pi via a UDP packet. These are encoded in a JSON string to keep the size low:

{'T': 1011, 'E': 0, 'R': 0, 'C': 1011, 'S': 0, 'N': 0, 'F': 41, 'W': 23, 'I': 0, 'P': 947}

UDP was chosen for a few reasons:

Data loss is unlikely as both devices are connected to the same router.

It's okay if a couple packets are lost or arrive out-of-order, as the next packet always contains the full dataset needed to update the display.

UDP connections are quicker to establish - no three-way handshake needed. Any network latency will not slow down my PHPUnit listener (and thus slow down test execution).

LED Display

On the Raspberry Pi, I have a simple Python script listening for these UDP packets and updating ("redrawing") the Unicorn pHAT LED display:

 #!/usr/bin/env python 
import json 
import socket 
import time 
import unicornhat as unicorn 
UDP_IP = "0.0.0.0" 
UDP_PORT = 5005 
# Define constants for each JSON key 
NO_RESULT = 'N' 
ERROR = 'E' 
FAILURE = 'F' 
WARNING = 'W' 
INCOMPLETE = 'I' 
RISKY = 'R' 
SKIPPED = 'S' 
PASSED = 'P' 
TOTAL = 'T' 
COMPLETED = 'C' 
TOTAL_PIXELS = 4*8 
unicorn.set_layout(unicorn.PHAT) 
unicorn.rotation(0) 
unicorn.brightness(0.5) 
sock = socket.socket(socket.AF_INET, # Internet 
                    socket.SOCK_DGRAM) # UDP 
sock.bind((UDP_IP, UDP_PORT)) 
time.sleep(1) 
def set_pixel(i, r, g, b): 
   # Don't render more pixels than we have 
   if i > TOTAL_PIXELS - 1: 
       return 
   # Determine the X/Y coordinates based on pixel index 
   x = int(i % 8) 
   y = int((i - x) / 8) 
   unicorn.set_pixel(x, y, r, g, b) 
while True: 
   data, addr = sock.recvfrom(1024) 
   unicorn.set_all(0, 0, 0) 
   data = data.decode("utf-8") 
   if data == "reset": 
       unicorn.show() 
       continue 
   data = json.loads(data) 
   original_data = data.copy() 
   test_count = data[TOTAL] 
   # how many tests count towards a pixel? 
   tests_per_pixel = test_count / TOTAL_PIXELS 
   for i in range(0, TOTAL_PIXELS): 
       # Render test results in order of severity 
       if data[ERROR] > 0: 
           data[ERROR] -= tests_per_pixel 
           set_pixel(i, 255, 0, 0) 
           continue 
       elif data[FAILURE] > 0: 
           data[FAILURE] -= tests_per_pixel 
           set_pixel(i, 200, 0, 0) 
           continue 
       elif data[WARNING] > 0: 
           data[WARNING] -= tests_per_pixel 
           set_pixel(i, 255, 255, 0) 
           continue 
       elif data[INCOMPLETE] > 0: 
           data[INCOMPLETE] -= tests_per_pixel 
           set_pixel(i, 170, 170, 0) 
           continue 
       elif data[RISKY] > 0: 
           data[RISKY] -= tests_per_pixel 
           set_pixel(i, 255, 255, 0) 
           continue 
       elif data[SKIPPED] > 0: 
           data[SKIPPED] -= tests_per_pixel 
           set_pixel(i, 64, 64, 64) 
           continue 
       elif data[PASSED] > 0: 
           data[PASSED] -= tests_per_pixel 
           set_pixel(i, 0, 255, 0) 
           continue 
   # Update the display once the pixel buffers are set 
   unicorn.show()

The display only has 32 "pixels" (4 rows of 8 LEDs) which almost never matches the number of tests being run. As a result, each pixel needs to represent several test runs, so I simply divide the number of tests by the number of pixels to figure that out. For example, if I have 640 tests in my suite, each pixel represents 20 different tests.

Once I have that tests-per-pixel number, I'm looping through each pixel to see what color to light it. I'm using the following logic:

Determine the next-highest severity which has un-displayed results

Light the next LED with the corresponding color

Subtract the tests-per-pixel number from that severity level, giving us how many other results of that outcome need to be shown

Repeat

This approach ensures that failures will always be shown, even if I only have one failure out of 640 tests. As you'll see in the next section, this approach has its downsides, but it's simple enough to implement and communicates the results just fine.

Next Steps

I'd eventually like to refactor how data is sent to the Pi and "drawn" on the LED display. The current implementation receives the current totals and draws them in order of severity - failures will always be shown on the top left, even if they occurred towards the end of the test suite.

Ideally I'd like the errors to appear relative to when they occur in the test suite - for example, if the first errors appear 50% of the way through, I'd like the screen to show two full rows of green lights with the first red lights appearing on row 3.

This will require changing how I collect and send the data to the Pi. Instead of sending totals of each result type and using that to redraw the display, I'd need to provide a stream of results like PHPUnit does:

...............................FF..........

To help keep the data packets small, I'd encode each result using 2 bits instead of ASCII characters:

00 - Incomplete

01 - Passed

10 - Warning, skipped, risky, etc.

11 - Error or failure

(A more clever encoding scheme could include some type of compression to condense a long sequence of identical outcomes into fewer bytes).

On the Pi, I'd decode that information and use it to light the pixels accordingly.

<?php

namespace ColinODell\PHPUnicorn;

use Exception;
use PHPUnit_Framework_AssertionFailedError;
use PHPUnit_Framework_Test;
use PHPUnit_Framework_TestSuite;
use PHPUnit_Framework_Warning;

class PHPUnicornListener extends \PHPUnit_Framework_BaseTestListener
{
    const NO_RESULT = 'N';
    const ERROR = 'E';
    const FAILURE = 'F';
    const INCOMPLETE = 'I';
    const RISKY = 'R';
    const SKIPPED = 'S';
    const PASSED = 'P';
    const WARNING = 'W';
    const TOTAL = 'T';
    const COMPLETED = 'C';

    private $currentTestPassed = false;
    private $counts = [];

    /**
     * @var string
     */
    private $host;

    /**
     * @var int
     */
    private $port;

    /**
     * @var resource
     */
    private $socket;

    /**
     * @param string $host
     * @param int    $port
     */
    public function __construct($host, $port)
    {
        $this->host = $host;
        $this->port = $port;
        $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
        $this->resetCounts();
        $this->broadcast();
    }

    /**
     * Ensure socket is closed
     */
    public function __destruct()
    {
        socket_close($this->socket);
        $this->socket = null;
    }

    /**
     * {@inheritdoc}
     */
    public function addError(PHPUnit_Framework_Test $test, Exception $e, $time)
    {
        $this->counts[self::ERROR]++;
        $this->currentTestPassed = false;
    }

    /**
     * {@inheritdoc}
     */
    public function addFailure(PHPUnit_Framework_Test $test, PHPUnit_Framework_AssertionFailedError $e, $time) {
        $this->counts[self::FAILURE]++;
        $this->currentTestPassed = false;
    }

    /**
     * {@inheritdoc}
     */
    public function addWarning(PHPUnit_Framework_Test $test, PHPUnit_Framework_Warning $e, $time)
    {
        $this->counts[self::WARNING]++;
        $this->currentTestPassed = false;
    }


    /**
     * {@inheritdoc}
     */
    public function addIncompleteTest(PHPUnit_Framework_Test $test, Exception $e, $time) {
        $this->counts[self::INCOMPLETE]++;
        $this->currentTestPassed = false;
    }

    /**
     * {@inheritdoc}
     */
    public function addRiskyTest(PHPUnit_Framework_Test $test, Exception $e, $time) {
        $this->counts[self::RISKY]++;
        $this->currentTestPassed = false;
    }

    /**
     * {@inheritdoc}
     */
    public function addSkippedTest(PHPUnit_Framework_Test $test, Exception $e, $time)
    {
        $this->counts[self::SKIPPED]++;
        $this->currentTestPassed = false;
    }

    /**
     * A test suite has started.
     *
     * {@inheritdoc}
     */
    public function startTestSuite(PHPUnit_Framework_TestSuite $suite)
    {
        // Test suites can contain child test suites.  This function is always
        // called with the top-most parent, so use its count method to determine
        // how many tests there are (it'll count all sub-children recursively).
        if ($this->counts[self::TOTAL] == 0) {
            $this->counts[self::TOTAL] = $suite->count();
        }
    }

    /**
     * A test suite ended.
     *
     * {@inheritdoc}
     */
    public function endTestSuite(PHPUnit_Framework_TestSuite $suite)
    {
        // Send the results over to the Pi
        $this->broadcast();
    }

    /**
     * A single test started.
     *
     * {@inheritdoc}
     */
    public function startTest(PHPUnit_Framework_Test $test)
    {
        // There's no method like addPassed(), so we'll assume the test passes
        // unless one of the other add___() methods are called.
        $this->currentTestPassed = true;
    }

    /**
     * A test ended.
     *
     * {@inheritdoc}
     */
    public function endTest(PHPUnit_Framework_Test $test, $time)
    {
        if ($this->currentTestPassed) {
            $this->counts[self::PASSED]++;
        }

        $this->counts[self::COMPLETED]++;

        $this->broadcast();
    }

    private function resetCounts()
    {
        $this->counts = [
            self::NO_RESULT => 0,
            self::ERROR => 0,
            self::FAILURE => 0,
            self::WARNING => 0,
            self::INCOMPLETE => 0,
            self::RISKY => 0,
            self::SKIPPED => 0,
            self::PASSED => 0,
            self::TOTAL => 0,
            self::COMPLETED => 0,
        ];
    }

    private function broadcast()
    {
        if ($this->counts[self::TOTAL] == 0) {
            // Ask the Pi to clear the screen when we first start
            $message = 'reset';
        } else {
            $message = json_encode($this->counts);
        }

        socket_sendto($this->socket, $message, strlen($message), 0, $this->host, $this->port);
    }
}

#!/usr/bin/env python

import json
import socket
import time
import unicornhat as unicorn

UDP_IP = "0.0.0.0"
UDP_PORT = 5005

# Define constants for each JSON key
NO_RESULT = 'N'
ERROR = 'E'
FAILURE = 'F'
WARNING = 'W'
INCOMPLETE = 'I'
RISKY = 'R'
SKIPPED = 'S'
PASSED = 'P'
TOTAL = 'T'
COMPLETED = 'C'

TOTAL_PIXELS = 4*8


unicorn.set_layout(unicorn.PHAT)
unicorn.rotation(0)
unicorn.brightness(0.5)

sock = socket.socket(socket.AF_INET, # Internet
                     socket.SOCK_DGRAM) # UDP
sock.bind((UDP_IP, UDP_PORT))

time.sleep(1)

def set_pixel(i, r, g, b):
    # Don't render more pixels than we have
    if i > TOTAL_PIXELS - 1:
        return

    # Determine the X/Y coordinates based on pixel index
    x = int(i % 8)
    y = int((i - x) / 8)

    unicorn.set_pixel(x, y, r, g, b)


while True:
    data, addr = sock.recvfrom(1024)

    unicorn.set_all(0, 0, 0)

    data = data.decode("utf-8")
    if data == "reset":
        unicorn.show()
        continue

    data = json.loads(data)
    original_data = data.copy()
    print(data)

    test_count = data[TOTAL]
    # how many tests count towards a pixel?
    tests_per_pixel = test_count / TOTAL_PIXELS

    for i in range(0, TOTAL_PIXELS):
        # Render test results in order of severity
        if data[ERROR] > 0:
            data[ERROR] -= tests_per_pixel
            set_pixel(i, 255, 0, 0)
            continue
        elif data[FAILURE] > 0:
            data[FAILURE] -= tests_per_pixel
            set_pixel(i, 200, 0, 0)
            continue
        elif data[WARNING] > 0:
            data[WARNING] -= tests_per_pixel
            set_pixel(i, 255, 255, 0)
            continue
        elif data[INCOMPLETE] > 0:
            data[INCOMPLETE] -= tests_per_pixel
            set_pixel(i, 170, 170, 0)
            continue
        elif data[RISKY] > 0:
            data[RISKY] -= tests_per_pixel
            set_pixel(i, 255, 255, 0)
            continue
        elif data[SKIPPED] > 0:
            data[SKIPPED] -= tests_per_pixel
            set_pixel(i, 64, 64, 64)
            continue
        elif data[PASSED] > 0:
            data[PASSED] -= tests_per_pixel
            set_pixel(i, 0, 255, 0)
            continue

    # Update the display once the pixel buffers are set
    unicorn.show()

Credits

Colin O'Dell

1 project • 43 followers

Lead Web Developer at Unleashed Technologies. Author of league/commonmark. Conference speaker. Arduino enthusiast.

PHPUnicorn - Visualizing PHPUnit Tests

Things used in this project

Hardware components

Software apps and online services

Hand tools and fabrication machines

Story

Goal

Hardware & Assembly

Extending PHPUnit

LED Display

Next Steps

Code

PHPUnicornListener.php

display.py

Credits

Colin O'Dell

Comments

Embed the widget on your own site

PHPUnicorn - Visualizing PHPUnit Tests

PHPUnicorn - Visualizing PHPUnit Tests

Things used in this project

Hardware components

Software apps and online services

Hand tools and fabrication machines

Story

Goal

Hardware & Assembly

Extending PHPUnit

LED Display

Next Steps

Code

PHPUnicornListener.php

display.py

Credits

Colin O'Dell

Comments

Related channels and tags