text: /* Copyright 2013-present Barefoot Networks, Inc.
   *
   * 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
   *
   *   http://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.
   */
  
  #include <gtest/gtest.h>
  
  #include <bm/config.h>
  #include <bm/bm_sim/dev_mgr.h>
  #include <bm/bm_sim/port_monitor.h>
  #include <bm/bm_apps/packet_pipe.h>
  
  #include <algorithm>
  #include <cassert>
  #include <chrono>
  #include <iostream>
  #include <map>
  #include <unordered_map>
  #include <thread>
  #include <mutex>
  #include <condition_variable>
  #include <string>
  #include <vector>
  
  #include "utils.h"
  
  using namespace bm;
  
  using testing::Types;
  
  class TestDevMgrImp : public DevMgrIface {
   public:
    TestDevMgrImp() {
      // 0 is device_id
      p_monitor = PortMonitorIface::make_active(0);
    }
  
    // Not part of public interface, just for testing
    void set_port_status(port_t port, const PortStatus status) {
      ASSERT_TRUE(status == PortStatus::PORT_UP ||
                  status == PortStatus::PORT_DOWN);
      std::lock_guard<std::mutex> lock(status_mutex);
      auto it = port_status.find(port);
      bool exists = it != port_status.end();
      ASSERT_TRUE(exists) << "Incorrect op sequence on port: " << port
                          << std::endl;
      if ((it->second == PortStatus::PORT_ADDED ||
           it->second == PortStatus::PORT_DOWN) &&
          status == PortStatus::PORT_UP) {
        port_status[port] = PortStatus::PORT_UP;
      } else if ((it->second == PortStatus::PORT_ADDED ||
                  it->second == PortStatus::PORT_UP) &&
                 status == PortStatus::PORT_DOWN) {
          port_status[port] = PortStatus::PORT_DOWN;
      } else {
        ASSERT_TRUE(false) << "Incorrect op sequence on port: " << port
                           << std::endl;
      }
    }
  
   private:
    bool port_is_up_(port_t port) const override {
      std::lock_guard<std::mutex> lock(status_mutex);
      auto it = port_status.find(port);
      bool exists = (it != port_status.end());
  
      return (exists && ((it->second == PortStatus::PORT_ADDED) ||
                         (it->second == PortStatus::PORT_UP)));
    }
  
    std::map<port_t, PortInfo> get_port_info_() const override {
      // TODO(antonin): add test
      return {};
    }
  
    ReturnCode port_add_(const std::string &iface_name, port_t port_num,
                         const PortExtras &port_extras) override {
      (void) iface_name;
      (void) port_extras;
      std::lock_guard<std::mutex> lock(status_mutex);
      auto it = port_status.find(port_num);
      if (it != port_status.end()) return ReturnCode::ERROR;
      port_status.insert(std::make_pair(port_num, PortStatus::PORT_ADDED));
      return ReturnCode::SUCCESS;
    }
  
    ReturnCode port_remove_(port_t port_num) override {
      std::lock_guard<std::mutex> lock(status_mutex);
      auto it = port_status.find(port_num);
      if (it == port_status.end()) return ReturnCode::ERROR;
      port_status.erase(it);
      return ReturnCode::SUCCESS;
    }
  
    ReturnCode set_packet_handler_(const PacketHandler &handler,
                                   void *cookie) override {
      (void) handler;
      (void) cookie;
      return ReturnCode::SUCCESS;
    }
  
    void transmit_fn_(DevMgrIface::port_t port_num, const char *buffer, int len)
    override {
      (void)port_num;
      (void)buffer;
      (void)len;
    }
  
    void start_() override {}
  
    std::map<port_t, PortStatus> port_status{};
    mutable std::mutex status_mutex{};
  };
  
  class DevMgrTest : public ::testing::Test {
   public:
    void port_status(DevMgrIface::port_t port_num,
                     const DevMgrIface::PortStatus status) {
      (void)port_num;
      std::lock_guard<std::mutex> lock(cnt_mutex);
      cb_counts[status]++;
    }
  
   protected:
    DevMgrTest()
        : g_mgr(new TestDevMgrImp()) {
      g_mgr->start();
    }
  
    void register_callback() {
      port_cb = std::bind(&DevMgrTest::port_status, this, std::placeholders::_1,
                          std::placeholders::_2);
      g_mgr->register_status_cb(DevMgrIface::PortStatus::PORT_ADDED, port_cb);
      g_mgr->register_status_cb(DevMgrIface::PortStatus::PORT_REMOVED, port_cb);
      g_mgr->register_status_cb(DevMgrIface::PortStatus::PORT_UP, port_cb);
      g_mgr->register_status_cb(DevMgrIface::PortStatus::PORT_DOWN, port_cb);
    }
  
    void reset_counts() {
      std::lock_guard<std::mutex> lock(cnt_mutex);
      for (auto &kv : cb_counts) {
        kv.second = 0;
      }
    }
  
    uint32_t get_count(DevMgrIface::PortStatus status) {
      std::lock_guard<std::mutex> lock(cnt_mutex);
      return cb_counts.at(status);
    }
  
    DevMgrIface::PortStatusCb port_cb{};
    std::map<DevMgrIface::PortStatus, uint32_t> cb_counts{};
    mutable std::mutex cnt_mutex{};
    std::unique_ptr<TestDevMgrImp> g_mgr{nullptr};
  };
  
  TEST_F(DevMgrTest, cb_test) {
    const unsigned int NPORTS = 10;
  
    register_callback();
  
    for (unsigned int i = 0; i < NPORTS; i++) {
      g_mgr->port_add("dummyport", i, {});
    }
    std::this_thread::sleep_for(std::chrono::seconds(2));
    ASSERT_EQ(NPORTS, get_count(DevMgrIface::PortStatus::PORT_ADDED))
        << "Port add callbacks incorrect" << std::endl;
    reset_counts();
  
    for (unsigned int i = 0; i < NPORTS; i++) {
      g_mgr->set_port_status(i, DevMgrIface::PortStatus::PORT_DOWN);
    }
    std::this_thread::sleep_for(std::chrono::seconds(2));
    ASSERT_EQ(NPORTS, get_count(DevMgrIface::PortStatus::PORT_DOWN))
        << "Port down callbacks incorrect" << std::endl;
  
    for (unsigned int i = 0; i < NPORTS; i++) {
      g_mgr->set_port_status(i, DevMgrIface::PortStatus::PORT_UP);
    }
    std::this_thread::sleep_for(std::chrono::seconds(2));
    ASSERT_EQ(NPORTS, get_count(DevMgrIface::PortStatus::PORT_UP))
        << "Port up callbacks incorrect" << std::endl;
  
    for (unsigned int i = 0; i < NPORTS; i++) {
      g_mgr->port_remove(i);
    }
    std::this_thread::sleep_for(std::chrono::seconds(2));
    ASSERT_EQ(NPORTS, get_count(DevMgrIface::PortStatus::PORT_REMOVED))
        << "Number of port remove callbacks incorrect" << std::endl;
  }
  
  class PacketInReceiver {
   public:
    enum class Status { CAN_READ, CAN_RECEIVE };
  
    PacketInReceiver(size_t max_size)
        : max_size(max_size) {
      buffer_.reserve(max_size);
    }
  
    void receive(int port_num, const char *buffer, int len, void *cookie) {
      (void) cookie;
      if (static_cast<size_t>(len) > max_size) return;
      std::unique_lock<std::mutex> lock(mutex);
      while (status != Status::CAN_RECEIVE) {
        can_receive.wait(lock);
      }
      buffer_.insert(buffer_.end(), buffer, buffer + len);
      port = port_num;
      status = Status::CAN_READ;
      can_read.notify_one();
    }
  
    bool read(char *dst, size_t len, int *recv_port,
              unsigned int timeout_ms = 1000) {
      len = (len > max_size) ? max_size : len;
      using clock = std::chrono::system_clock;
      clock::time_point tp_start = clock::now();
      clock::time_point tp_end = tp_start + std::chrono::milliseconds(timeout_ms);
      std::unique_lock<std::mutex> lock(mutex);
      while (status != Status::CAN_READ) {
        if (clock::now() > tp_end)
          return false;
        can_read.wait_until(lock, tp_end);
      }
      std::copy(buffer_.begin(), buffer_.begin() + len, dst);
      buffer_.clear();
      *recv_port = port;
      status = Status::CAN_RECEIVE;
      can_receive.notify_one();
      return true;
    }
  
    Status check_status() {
      std::unique_lock<std::mutex> lock(mutex);
      return status;
    }
  
   private:
    size_t max_size;
    std::vector<char> buffer_;
    int port;
    Status status{Status::CAN_RECEIVE};
    mutable std::mutex mutex{};
    mutable std::condition_variable can_receive{};
    mutable std::condition_variable can_read{};
  };
  
  #ifdef BM_NANOMSG_ON
  
  // is here because DevMgr has a protected destructor
  class PacketInSwitch : public DevMgr { };
  
  class PacketInDevMgrTest : public ::testing::Test {
   protected:
    static constexpr size_t kMaxBufferSize = 512;
  
    PacketInDevMgrTest()
        : packet_inject(addr) { }
  
    void SetUp_(bool enforce_ports,
                std::shared_ptr<TransportIface> notifications_transport) {
      // 0 is device id
      sw.set_dev_mgr_packet_in(0, addr, notifications_transport, enforce_ports);
      sw.start();
      auto cb_switch = std::bind(&PacketInReceiver::receive, &recv_switch,
                                 std::placeholders::_1, std::placeholders::_2,
                                 std::placeholders::_3, std::placeholders::_4);
      sw.set_packet_handler(cb_switch, nullptr);
  
      packet_inject.start();
      auto cb_lib = std::bind(&PacketInReceiver::receive, &recv_lib,
                              std::placeholders::_1, std::placeholders::_2,
                              std::placeholders::_3, std::placeholders::_4);
      packet_inject.set_packet_receiver(cb_lib, nullptr);
    }
  
    virtual void SetUp() {
      SetUp_(false, nullptr);
    }
  
    virtual void TearDown() {
    }
  
    bool check_recv(PacketInReceiver *receiver,
                    int send_port, const char *send_buffer, size_t size,
                    unsigned int timeout_ms = 1000) {
      char recv_buffer[kMaxBufferSize];
      memset(recv_buffer, 0, sizeof(recv_buffer));
      if (size > sizeof(recv_buffer)) return false;
      int recv_port = -1;
      if (!receiver->read(recv_buffer, size, &recv_port, timeout_ms))
        return false;
      if (recv_port != send_port) return false;
      return !memcmp(recv_buffer, send_buffer, size);
    }
  
    const std::string addr = "ipc:///tmp/test_packet_in_abc123";
  
    PacketInReceiver recv_switch{kMaxBufferSize};
    PacketInReceiver recv_lib{kMaxBufferSize};
  
    PacketInSwitch sw;
  
    bm_apps::PacketInject packet_inject;
  };
  
  TEST_F(PacketInDevMgrTest, PacketInTest) {
    constexpr int port = 2;
    const char pkt[] = {'\x0a', '\xba'};
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_switch.check_status());
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_lib.check_status());
    // switch -> lib
    sw.transmit_fn(port, pkt, sizeof(pkt));
    ASSERT_TRUE(check_recv(&recv_lib, port, pkt, sizeof(pkt)));
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_switch.check_status());
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_lib.check_status());
    // lib -> switch
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_TRUE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
  }
  
  TEST_F(PacketInDevMgrTest, InfoRequestTest) {
    constexpr int port = 2;
    // using info_type 0 and 1, but can be any integer value at the moment
    // we expect a return value of 1 (error) as bmv2 does not support any request
    // type
    std::string rep_v;
    ASSERT_EQ(1, packet_inject.request_info(port, 0, &rep_v));
    ASSERT_EQ(1, packet_inject.request_info(port, 1, &rep_v));
  }
  
  class PacketInDevMgrPortStatusTest : public PacketInDevMgrTest {
   protected:
    PacketInDevMgrPortStatusTest() { }
  
    virtual void SetUp() {
      SetUp_(true, nullptr);
      register_callback();
    }
  
    virtual void TearDown() {
    }
  
    void port_status(DevMgrIface::port_t port_num,
                     const DevMgrIface::PortStatus status) {
      (void)port_num;
      std::lock_guard<std::mutex> lock(cnt_mutex);
      cb_counts[status]++;
    }
  
    void register_callback() {
      port_cb = std::bind(&PacketInDevMgrPortStatusTest::port_status, this,
                          std::placeholders::_1, std::placeholders::_2);
      sw.register_status_cb(DevMgrIface::PortStatus::PORT_ADDED, port_cb);
      sw.register_status_cb(DevMgrIface::PortStatus::PORT_REMOVED, port_cb);
      sw.register_status_cb(DevMgrIface::PortStatus::PORT_UP, port_cb);
      sw.register_status_cb(DevMgrIface::PortStatus::PORT_DOWN, port_cb);
    }
  
    void reset_counts() {
      std::lock_guard<std::mutex> lock(cnt_mutex);
      for (auto &kv : cb_counts) {
        kv.second = 0;
      }
    }
  
    uint32_t get_count(DevMgrIface::PortStatus status) {
      std::lock_guard<std::mutex> lock(cnt_mutex);
      return cb_counts[status];
    }
  
    void check_and_reset_counts(uint32_t count_added, uint32_t count_removed,
                                uint32_t count_up, uint32_t count_down) {
      ASSERT_EQ(count_added, get_count(DevMgrIface::PortStatus::PORT_ADDED));
      ASSERT_EQ(count_removed, get_count(DevMgrIface::PortStatus::PORT_REMOVED));
      ASSERT_EQ(count_up, get_count(DevMgrIface::PortStatus::PORT_UP));
      ASSERT_EQ(count_down, get_count(DevMgrIface::PortStatus::PORT_DOWN));
      reset_counts();
    }
  
    DevMgrIface::PortStatusCb port_cb{};
    std::map<DevMgrIface::PortStatus, uint32_t> cb_counts{};
    mutable std::mutex cnt_mutex{};
  };
  
  TEST_F(PacketInDevMgrPortStatusTest, BadPort) {
    constexpr int port = 2;
    const char pkt[] = {'\x0a', '\xba'};
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_switch.check_status());
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_lib.check_status());
    // lib -> switch
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_FALSE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
  }
  
  TEST_F(PacketInDevMgrPortStatusTest, Basic) {
    constexpr int port = 2;
    const char pkt[] = {'\x0a', '\xba'};
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_switch.check_status());
    ASSERT_EQ(PacketInReceiver::Status::CAN_RECEIVE, recv_lib.check_status());
    packet_inject.port_add(port);
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_TRUE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
    packet_inject.port_remove(port);
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_FALSE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
  }
  
  TEST_F(PacketInDevMgrPortStatusTest, Status) {
    constexpr int port = 2;
    const char pkt[] = {'\x0a', '\xba'};
  
    packet_inject.port_add(port);
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_TRUE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    check_and_reset_counts(1u, 0u, 1u, 0u);
  
    packet_inject.port_bring_down(port);
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_FALSE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    check_and_reset_counts(0u, 0u, 0u, 1u);
  
    packet_inject.port_bring_up(port);
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_TRUE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    check_and_reset_counts(0u, 0u, 1u, 0u);
  
    packet_inject.port_remove(port);
    packet_inject.send(port, pkt, sizeof(pkt));
    ASSERT_FALSE(check_recv(&recv_switch, port, pkt, sizeof(pkt)));
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    check_and_reset_counts(0u, 1u, 0u, 0u);
  }
  
  #endif  // BM_NANOMSG_ON
  
  struct PMActive { };
  struct PMPassive { };
  
  // testing the port status nanomsg notifications
  template <typename PMType>
  class PortMonitorTest : public ::testing::Test {
   protected:
    using port_t = DevMgrIface::port_t;
    using PortStatus = DevMgrIface::PortStatus;
  
    static constexpr device_id_t device_id = 0;
  
    PortMonitorTest()
        : notifications_writer(new MemoryAccessor(4096)) {
      make_monitor();
    }
  
    virtual void SetUp() {
      auto status_fn = std::bind(&PortMonitorTest::port_is_up,
                                 this, std::placeholders::_1);
      p_monitor->start(status_fn);
    }
  
    // virtual void TearDown() { }
  
    void make_monitor();
  
    bool port_is_up(port_t port);
  
    void set_port_status(port_t port, PortStatus status);
  
    const PortMonitorIface::msg_hdr_t *get_msg_hdr(const char *buffer) const {
      return reinterpret_cast<const PortMonitorIface::msg_hdr_t *>(buffer);
    }
  
    const PortMonitorIface::one_status_t *get_statuses(const char *buffer) const {
      return reinterpret_cast<const PortMonitorIface::one_status_t *>(
          buffer + sizeof(PortMonitorIface::msg_hdr_t));
    }
  
    std::unordered_map<port_t, PortStatus> port_status{};
    mutable std::mutex status_mutex{};
    std::unique_ptr<PortMonitorIface> p_monitor{nullptr};
    std::shared_ptr<MemoryAccessor> notifications_writer{nullptr};
  };
  
  template <typename PMType>
  constexpr device_id_t PortMonitorTest<PMType>::device_id;
  
  template<>
  void
  PortMonitorTest<PMPassive>::make_monitor() {
    p_monitor = PortMonitorIface::make_passive(device_id, notifications_writer);
  }
  
  template<>
  void
  PortMonitorTest<PMActive>::make_monitor() {
    p_monitor = PortMonitorIface::make_active(device_id, notifications_writer);
  }
  
  template<>
  bool
  PortMonitorTest<PMPassive>::port_is_up(port_t port) {
    (void)port;
    assert(0);
    return false;
  }
  
  template<>
  bool
  PortMonitorTest<PMActive>::port_is_up(port_t port) {
    std::lock_guard<std::mutex> lock(status_mutex);
    return (port_status.at(port) == PortStatus::PORT_UP);
  }
  
  template<>
  void
  PortMonitorTest<PMPassive>::set_port_status(port_t port, PortStatus status) {
    p_monitor->notify(port, status);
  }
  
  template<>
  void
  PortMonitorTest<PMActive>::set_port_status(port_t port, PortStatus status) {
    if (status == PortStatus::PORT_ADDED || status == PortStatus::PORT_UP) {
      std::lock_guard<std::mutex> lock(status_mutex);
      port_status[port] = PortStatus::PORT_UP;
    }
    if (status == PortStatus::PORT_REMOVED || status == PortStatus::PORT_DOWN) {
      std::lock_guard<std::mutex> lock(status_mutex);
      port_status[port] = PortStatus::PORT_DOWN;
    }
    if (status == PortStatus::PORT_ADDED || status == PortStatus::PORT_REMOVED) {
      p_monitor->notify(port, status);
    }
  }
  
  using PMTypes = Types<PMPassive, PMActive>;
  
  TYPED_TEST_CASE(PortMonitorTest, PMTypes);
  
  using std::chrono::milliseconds;
  using std::this_thread::sleep_for;
  
  TYPED_TEST(PortMonitorTest, Basic) {
    using PortStatus = DevMgrIface::PortStatus;
    using port_t = DevMgrIface::port_t;
    using Status = MemoryAccessor::Status;
    const port_t port = 1;
  
    char buffer[128];
  
    this->set_port_status(port, PortStatus::PORT_ADDED);
    this->set_port_status(port, PortStatus::PORT_UP);
    this->notifications_writer->read(buffer, sizeof(buffer));
    auto msg_hdr = this->get_msg_hdr(buffer);
    auto statuses = this->get_statuses(buffer);
    ASSERT_EQ(0, memcmp("PRT|", msg_hdr->sub_topic, sizeof(msg_hdr->sub_topic)));
    ASSERT_EQ(this->device_id, msg_hdr->switch_id);
    ASSERT_EQ(1u, msg_hdr->num_statuses);
    ASSERT_EQ(port, statuses[0].port);
    ASSERT_EQ(1, statuses[0].status);
  
    this->set_port_status(port, PortStatus::PORT_UP);
    sleep_for(milliseconds(300));
    ASSERT_NE(Status::CAN_READ, this->notifications_writer->check_status());
  
    this->set_port_status(port, PortStatus::PORT_DOWN);
    this->notifications_writer->read(buffer, sizeof(buffer));
    msg_hdr = this->get_msg_hdr(buffer);
    statuses = this->get_statuses(buffer);
    ASSERT_EQ(0, memcmp("PRT|", msg_hdr->sub_topic, sizeof(msg_hdr->sub_topic)));
    ASSERT_EQ(this->device_id, msg_hdr->switch_id);
    ASSERT_EQ(1u, msg_hdr->num_statuses);
    ASSERT_EQ(port, statuses[0].port);
    ASSERT_EQ(0, statuses[0].status);
  }
  