ptyを使用したシリアルデバイスドライバの単体テスト例

serial_driver_tester.cpp

#include <fcntl.h>
#include <termios.h>
#include <unistd.h>

#include <atomic>
#include <cerrno>
#include <chrono>
#include <cstdlib>
#include <cstring>
#include <functional>
#include <iostream>
#include <string>
#include <thread>

// 疑似的なデバイスを実装するクラス．
// この例では，受信データを2倍して送り返すだけのものを実装している．
class PseudoDevice {
 public:
  PseudoDevice()
      : fd_pty_(0), read_buffer_(), is_running_(false), read_thread_() {}
  ~PseudoDevice() {}

  void Start() {
    if (OpenDevice()) {
      is_running_ = true;
      read_thread_ = std::thread(std::bind(&PseudoDevice::ReadLoop, this));
    }
  }

  void Stop() {
    is_running_ = false;
    if (read_thread_.joinable()) {
      read_thread_.join();
    }
    CloseDevice();
  }

  std::string GetDeviceFileName() const {
    // ファイルディスクリプタからptyのファイル名を取得
    return std::string(ptsname(fd_pty_));
  }

 private:
  bool OpenDevice() {
    fd_pty_ = posix_openpt(O_RDWR);
    if (fd_pty_ < 0) {
      std::cerr << "Error while opening pseudo terminal: " << strerror(errno)
                << "(" << errno << ")" << std::endl;
      return false;
    }
    if (grantpt(fd_pty_) != 0) {
      std::cerr << "Error while grantpt: " << strerror(errno) << "(" << errno
                << ")" << std::endl;
      return false;
    }
    if (unlockpt(fd_pty_) != 0) {
      std::cerr << "Error while unlockpt: " << strerror(errno) << "(" << errno
                << ")" << std::endl;
      return false;
    }
    return true;
  }

  void CloseDevice() { close(fd_pty_); }

  void ReadLoop() {
    using namespace std::chrono_literals;
    while (is_running_) {
      // デバイスの応答遅れを模擬するディレイ．
      std::this_thread::sleep_for(2ms);
      const int received_bytes =
          read(fd_pty_, &read_buffer_, sizeof(read_buffer_));
      if (received_bytes > 0) {
        // この例では，受信したデータを2倍にして送り返すデバイスを実装する．
        for (int i = 0; i < received_bytes; i++) {
          const char send_data = static_cast<char>(read_buffer_[i] * 2);
          write(fd_pty_, &send_data, 1);

          std::cout << "[Device] Received: "
                    << static_cast<int>(read_buffer_[i]) << std::endl;
          std::cout << "[Device] Sent: " << static_cast<int>(send_data)
                    << std::endl;
        }
      }
    }
  }

  int fd_pty_;
  char read_buffer_[1024];
  std::atomic<bool> is_running_;
  std::thread read_thread_;
};

// シリアルデバイスとの通信を行うクラス．こいつがテスト対象．
class DeviceDriver {
 public:
  DeviceDriver(const std::string& device_file_name)
      : device_file_name_(device_file_name),
        fd_serial_port_(-1),
        read_buffer_(),
        received_data_(-1),
        is_running_(false),
        read_thread_() {}

  ~DeviceDriver(){};

  void Start() {
    if (OpenDevice()) {
      is_running_ = true;
      read_thread_ = std::thread(std::bind(&DeviceDriver::ReadLoop, this));
    }
  }

  void Stop() {
    is_running_ = false;
    if (read_thread_.joinable()) {
      read_thread_.join();
    }
    CloseDevice();
  }

  void SendDataToDevice(int data) {
    const char data_char = static_cast<char>(data);
    write(fd_serial_port_, &data_char, 1);
    std::cout << "[Worker] Sent: " << static_cast<int>(data_char) << std::endl;
  }

  int GetReceivedData() const { return received_data_; }

 private:
  bool OpenDevice() {
    fd_serial_port_ = open(device_file_name_.data(), O_RDWR);
    if (fd_serial_port_ < 0) {
      std::cerr << "File open failed: " << strerror(errno) << "(" << errno
                << ")" << std::endl;
      return false;
    }

    // シリアルデバイスの設定．ptyでこの辺をいじってもエラーは起きない．素晴らしい．
    struct termios tty;
    if (tcgetattr(fd_serial_port_, &tty) != 0) {
      std::cerr << "Error %i from tcgetattr: " << errno << strerror(errno)
                << std::endl;
    }
    tty.c_cflag &= ~(PARENB | CSTOPB | CRTSCTS);
    tty.c_cflag |= CS8 | CREAD | CLOCAL;
    tty.c_lflag &= ~(ICANON | ECHO | ECHOE | ECHONL | ISIG);
    tty.c_iflag &= ~(IGNBRK | BRKINT | PARMRK | ISTRIP | INLCR | IGNCR | ICRNL |
                     IXON | IXOFF | IXANY);
    tty.c_oflag &= ~(OPOST | ONLCR);
    tty.c_cc[VTIME] = 1;
    tty.c_cc[VMIN] = 0;
    cfsetispeed(&tty, B9600);
    cfsetospeed(&tty, B9600);
    if (tcsetattr(fd_serial_port_, TCSANOW, &tty) != 0) {
      std::cerr << "Error %i from tcsetattr: " << errno << strerror(errno)
                << std::endl;
    }
    return true;
  }

  void ReadLoop() {
    using namespace std::chrono_literals;
    while (is_running_) {
      const int received_bytes =
          read(fd_serial_port_, &read_buffer_, sizeof(read_buffer_));
      if (received_bytes > 0) {
        // 受け取ったパケットを処理する．通常はパケットをパースするコードを書くが，
        // この例では簡単のため，受け取った値をそのまま受信データとして保持する．
        for (int i = 0; i < received_bytes; i++) {
          std::cout << "[Worker] Received: "
                    << static_cast<int>(read_buffer_[i]) << std::endl;
          received_data_ = read_buffer_[i];
        }
      }
    }
  }

  void CloseDevice() const { close(fd_serial_port_); }

  const std::string device_file_name_;
  int fd_serial_port_;
  char read_buffer_[1024];
  int received_data_;
  std::atomic<bool> is_running_;
  std::thread read_thread_;
};

int main(int, char**) {
  using namespace std::chrono_literals;

  constexpr int send_data = 2;
  constexpr int expected_data = send_data * 2;

  PseudoDevice device;
  device.Start();
  DeviceDriver driver(device.GetDeviceFileName());
  driver.Start();

  // データを送った後しばらくスリープしてから受信データを確認する
  driver.SendDataToDevice(send_data);
#if 1
  std::this_thread::sleep_for(10ms);
#else
  std::this_thread::sleep_for(1ms);  // ウェイトを短くすると失敗する
#endif
  const int actual_data = driver.GetReceivedData();

  driver.Stop();
  device.Stop();

  // 結果の表示
  const bool result = actual_data == expected_data;
  std::cout << "Test result: " << (result ? "PASSED" : "FAILED") << std::endl;
  if (!result) {
    std::cout << "\tExpected: " << expected_data << std::endl;
    std::cout << "\tActual: " << actual_data << std::endl;
  }
}