lucicfg/testdata/misc/triggerer.star

load('@stdlib//internal/graph.star', 'graph')
load('@stdlib//internal/luci/common.star', 'builder_ref', 'keys', 'triggerer')


### Toy builder+poller model.


def builder(bucket, name, triggered_by=[], triggers=[]):
  k = keys.builder(bucket, name)
  graph.add_node(k)
  ref = builder_ref.add(k)
  trg = triggerer.add(k)
  for t in triggered_by:
    graph.add_edge(keys.triggerer(t), ref)
  for t in triggers:
    graph.add_edge(trg, keys.builder_ref(t))
  return k


def poller(bucket, name, triggers=[]):
  k = keys.gitiles_poller(bucket, name)
  graph.add_node(k)
  trg = triggerer.add(k)
  for t in triggers:
    graph.add_edge(trg, keys.builder_ref(t))
  return k


def targets(poller_key):
  return [n.key for n in triggerer.targets(graph.node(poller_key))]


### Test cases.


def test_triggers():
  p1 = poller('buck', 'p1', triggers=['buck1/b1', 'b2'])
  p2 = poller('buck', 'p2', triggers=['b2'])
  p3 = poller('buck', 'p3')
  b1 = builder('buck1', 'b1')
  b2 = builder('buck2', 'b2')
  __native__.graph().finalize()
  assert.eq(targets(p1), [b1, b2])
  assert.eq(targets(p2), [b2])
  assert.eq(targets(p3), [])


def test_triggered_by():
  p1 = poller('buck', 'p1')
  p2 = poller('buck', 'p2')
  b1 = builder('buck1', 'b1', triggered_by=['buck/p1'])
  b2 = builder('buck2', 'b2', triggered_by=['p1', 'p2'])
  __native__.graph().finalize()
  assert.eq(targets(p1), [b1, b2])
  assert.eq(targets(p2), [b2])


def test_both_directions_not_an_error():
  p1 = poller('buck', 'p1', triggers=['buck1/b1', 'b2'])
  b1 = builder('buck1', 'b1', triggered_by=['p1'])
  b2 = builder('buck2', 'b2', triggered_by=['buck/p1'])
  __native__.graph().finalize()
  assert.eq(targets(p1), [b1, b2])


def test_ambiguous_builder_ref():
  p = poller('buck', 'p', triggers=['b'])
  b1 = builder('buck1', 'b')
  b2 = builder('buck2', 'b')
  __native__.graph().finalize()
  assert.fails(lambda: targets(p),
      'ambiguous reference "b" in core.gitiles_poller\("buck/p"\), possible variants:\n' +
      '  core.builder\("buck1/b"\)\n' +
      '  core.builder\("buck2/b"\)'
  )


def test_ambiguous_triggerer():
  p1 = poller('buck1', 'p')
  p2 = poller('buck2', 'p')
  bp = builder('buck3', 'p')
  b = builder('buck', 'b', triggered_by=['p'])
  __native__.graph().finalize()
  err_re = (
      'ambiguous reference "p" in core.builder\("buck/b"\), possible variants:\n' +
      '  core.gitiles_poller\("buck1/p"\)\n' +
      '  core.gitiles_poller\("buck2/p"\)\n' +
      '  core.builder\("buck3/p"\)'
  )
  assert.fails(lambda: targets(p1), err_re)
  assert.fails(lambda: targets(p2), err_re)


def test_triggering_cycle():
  builder('buck', 'b1', triggers=['b2'])
  assert.fails(
      lambda: builder('buck', 'b2', triggers=['b1']), 'introduces a cycle')


### Test runner.


def with_clean_state(cbs):
  for cb in cbs:
    __native__.clear_state()
    __native__.fail_on_errors()
    cb()


with_clean_state([
    test_triggers,
    test_triggered_by,
    test_both_directions_not_an_error,
    test_ambiguous_builder_ref,
    test_ambiguous_triggerer,
    test_triggering_cycle,
])