tests/testthat/test-4-subscriptions.R

In googlePubsubR: R Interface for Google 'Cloud Pub/Sub' REST API

test_that("Gets and lists subscriptions", {
  skip_on_cran()
  res_get <- subscriptions_get(sub_name)
  res_list <- subscriptions_list()
  res_exist <- subscriptions_exists(sub_name)
  res_exist_fail <- subscriptions_exists("junk-name")

  expect_s3_class(res_get, "Subscription")
  expect_type(res_list, "list")
  expect_true(res_exist)
  expect_false(res_exist_fail)
})

test_that("Subscription gets patched", {
  skip_on_cran()
  res <- subscriptions_patch(
    subscription = sub_name,
    topic = topic_name,
    msg_retention_duration = 2400,
    labels = list(new_a = "a", new_b = "b")
  )

  expect_s3_class(res, "Subscription")
  expect_equal(res$messageRetentionDuration, "2400s")
  expect_equal(res$labels$new_a, "a")
  expect_equal(res$labels$new_b, "b")
})

test_that("Pulls messages from a subscription", {
  skip_on_cran()
  topics_publish(msg, topic_name)
  
  msgs <- subscriptions_pull(sub_name)

  expect_s3_class(msgs$receivedMessages, "data.frame")
  expect_true(length(msgs$receivedMessages) > 0)
})

test_that("Aknowledges messages", {
  skip_on_cran()
  msgs <- subscriptions_pull(sub_name)
  res <- subscriptions_ack(msgs$receivedMessages$ackId, sub_name)

  expect_true(res)
})

test_that("Modifies the ack deadline ", {
  skip_on_cran()
  # FIXME: this test fails in randomly (more often in CI) with a 404 on the subscription. 
  # I suspect this might be due to some race conditions generated by deleting/re-creating
  # the sub in "Subscription detaches" test. For the time being, changing test order seems
  # to have solved the problem.
  Sys.sleep(0.5)
  
  # Publish a message to the topic
  topics_publish(msg, topic_name)

  msgs <- subscriptions_pull(sub_name)
  res <- subscriptions_modify_ack_deadline(
    sub_name, msgs$receivedMessages$ackId, 400
  )

  expect_true(res)
})

test_that("Subscription seeks to a point in time", {
  skip_on_cran()
  
  # We can seek to a snapshot or a specific timestamp 
  time_seek <- subscriptions_seek(sub_name,
                                  time =  strftime(Sys.time() - 20, "%Y-%m-%dT%H:%M:%SZ"))
  snap_seek <- subscriptions_seek(sub_name, snapshot = snap_name)
  
  expect_true(time_seek)
  expect_true(snap_seek)
})

test_that("Update push configurations", {
  skip_on_cran()
  res <- subscriptions_modify_pushconf(
    sub_name, PushConfig(push_endpoint = "https://example.com/push")
  )
  
  expect_true(res)
})

test_that("Subscription detaches", {
  skip_on_cran()
  res <- subscriptions_detach(sub_name)
  
  expect_true(res)
  # It is not possible to reattach a subscription to a topic, hence we need to delete
  # and recreate it for further testing.
  subscriptions_delete(sub_name)
  subscriptions_create(name = sub_name, topic = topic_name)
})