Introduce POSet representation

Signed-off-by: Alexandre Terrasa <[email protected]>

lib/spoom.rb

@@ -15,6 +15,7 @@ class Error < StandardError; end
 require "spoom/file_collector"
 require "spoom/context"
 require "spoom/colors"
+require "spoom/poset"
 require "spoom/deadcode"
 require "spoom/sorbet"
 require "spoom/cli"

lib/spoom/poset.rb

@@ -0,0 +1,198 @@
+# typed: strict
+# frozen_string_literal: true
+
+module Spoom
+  # A Poset is a set of elements with a partial order relation.
+  #
+  # The partial order relation is a binary relation that is reflexive, antisymmetric, and transitive.
+  # It can be used to represent a hierarchy of classes or modules, the dependencies between gems, etc.
+  class Poset
+    extend T::Sig
+    extend T::Generic
+
+    class Error < Spoom::Error; end
+
+    E = type_member { { upper: Object } }
+
+    sig { void }
+    def initialize
+      @nodes = T.let({}, T::Hash[E, Element[E]])
+    end
+
+    # Get the POSet element for a given value
+    #
+    # Raises if the element is not found
+    sig { params(value: E).returns(Element[E]) }
+    def [](value)
+      poe = @nodes[value]
+      raise Error, "POSet::Element not found for #{value}" unless poe
+
+      poe
+    end
+
+    # Add a node to the POSet
+    sig { params(value: E).returns(Element[E]) }
+    def add_node(value)
+      poe = @nodes[value]
+      return poe if poe
+
+      @nodes[value] = Element[E].new(self, value)
+    end
+
+    # Is the given value a node in the POSet?
+    sig { params(value: E).returns(T::Boolean) }
+    def node?(value)
+      @nodes.key?(value)
+    end
+
+    # Add a direct edge from one element to another
+    #
+    # Transitive edges (transitive closure) are automatically computed.
+    # Adds the nodes if they don't exist.
+    # If the direct edge already exists, nothing is done.
+    sig { params(from: E, to: E).void }
+    def add_direct_edge(from, to)
+      from_poe = add_node(from)
+      to_poe = add_node(to)
+
+      # We already added this direct edge, which means we already computed the transitive closure
+      return if from_poe.parents.include?(to)
+
+      # Add the direct edges
+      from_poe.dtos << to_poe
+      to_poe.dfroms << from_poe
+
+      # Compute the transitive closure
+
+      from_poe.tos << to_poe
+      from_poe.froms.each do |child_poe|
+        child_poe.tos << to_poe
+        to_poe.froms << child_poe
+
+        to_poe.tos.each do |parent_poe|
+          parent_poe.froms << child_poe
+          child_poe.tos << parent_poe
+        end
+      end
+
+      to_poe.froms << from_poe
+      to_poe.tos.each do |parent_poe|
+        parent_poe.froms << from_poe
+        from_poe.tos << parent_poe
+
+        from_poe.froms.each do |child_poe|
+          child_poe.tos << parent_poe
+          parent_poe.froms << child_poe
+        end
+      end
+    end
+
+    # Is there an edge (direct or indirect) from `from` to `to`?
+    sig { params(from: E, to: E).returns(T::Boolean) }
+    def edge?(from, to)
+      from_poe = @nodes[from]
+      return false unless from_poe
+
+      from_poe.ancestors.include?(to)
+    end
+
+    # Is there a direct edge from `from` to `to`?
+    sig { params(from: E, to: E).returns(T::Boolean) }
+    def direct_edge?(from, to)
+      self[from].parents.include?(to)
+    end
+
+    # Show the POSet as a DOT graph using xdot (used for debugging)
+    sig { params(direct: T::Boolean, transitive: T::Boolean).void }
+    def show_dot(direct: true, transitive: true)
+      Open3.popen3("xdot -") do |stdin, _stdout, _stderr, _thread|
+        stdin.write(to_dot(direct: direct, transitive: transitive))
+        stdin.close
+      end
+    end
+
+    # Return the POSet as a DOT graph
+    sig { params(direct: T::Boolean, transitive: T::Boolean).returns(String) }
+    def to_dot(direct: true, transitive: true)
+      dot = +"digraph {\n"
+      dot << "  rankdir=BT;\n"
+      @nodes.each do |element, poe|
+        dot << "  \"#{element}\";\n"
+        if direct
+          poe.parents.each do |to|
+            dot << "  \"#{element}\" -> \"#{to}\";\n"
+          end
+        end
+        if transitive # rubocop:disable Style/Next
+          poe.ancestors.each do |ancestor|
+            dot << "  \"#{element}\" -> \"#{ancestor}\" [style=dotted];\n"
+          end
+        end
+      end
+      dot << "}\n"
+    end
+
+    # An element in a POSet
+    class Element
+      extend T::Sig
+      extend T::Generic
+      include Comparable
+
+      E = type_member { { upper: Object } }
+
+      # The value held by this element
+      sig { returns(E) }
+      attr_reader :value
+
+      # Edges (direct and indirect) from this element to other elements in the same POSet
+      sig { returns(T::Set[Element[E]]) }
+      attr_reader :dtos, :tos, :dfroms, :froms
+
+      sig { params(poset: Poset[E], value: E).void }
+      def initialize(poset, value)
+        @poset = poset
+        @value = value
+        @dtos = T.let(Set.new, T::Set[Element[E]])
+        @tos = T.let(Set.new, T::Set[Element[E]])
+        @dfroms = T.let(Set.new, T::Set[Element[E]])
+        @froms = T.let(Set.new, T::Set[Element[E]])
+      end
+
+      sig { params(other: T.untyped).returns(T.nilable(Integer)) }
+      def <=>(other)
+        return unless other.is_a?(Element)
+        return 0 if self == other
+
+        if tos.include?(other)
+          -1
+        elsif froms.include?(other)
+          1
+        end
+      end
+
+      # Direct parents of this element
+      sig { returns(T::Array[E]) }
+      def parents
+        @dtos.map(&:value)
+      end
+
+      # Direct and indirect ancestors of this element
+      sig { returns(T::Array[E]) }
+      def ancestors
+        @tos.map(&:value)
+      end
+
+      # Direct children of this element
+      sig { returns(T::Array[E]) }
+      def children
+        @dfroms.map(&:value)
+      end
+
+      # Direct and indirect descendants of this element
+      sig { returns(T::Array[E]) }
+      def descendants
+        @froms.map(&:value)
+      end
+    end
+  end
+end

test/spoom/poset_test.rb

@@ -0,0 +1,189 @@
+# typed: true
+# frozen_string_literal: true
+
+require "test_helper"
+
+module Spoom
+  class PosetTest < Minitest::Test
+    def test_empty
+      poset = Poset[String].new
+
+      refute(poset.node?("A"))
+      refute(poset.edge?("A", "B"))
+    end
+
+    def test_add_node
+      poset = Poset[String].new
+
+      poset.add_node("A")
+      assert(poset.node?("A"))
+      refute(poset.node?("B"))
+
+      poset.add_node("B")
+      assert(poset.node?("A"))
+      assert(poset.node?("B"))
+    end
+
+    def test_add_edge_also_adds_nodes
+      poset = Poset[String].new
+      poset.add_direct_edge("A", "B")
+
+      assert(poset.node?("A"))
+      assert(poset.node?("B"))
+    end
+
+    def test_add_edge_creates_direct_edge
+      poset = Poset[String].new
+
+      poset.add_direct_edge("A", "B")
+      assert(poset.direct_edge?("A", "B"))
+
+      poset.add_direct_edge("B", "C")
+      assert(poset.direct_edge?("B", "C"))
+
+      refute(poset.direct_edge?("A", "C"))
+    end
+
+    def test_add_edge_creates_transitive_edge
+      poset = Poset[String].new
+
+      poset.add_direct_edge("A", "B")
+      assert(poset.edge?("A", "B"))
+
+      poset.add_direct_edge("B", "C")
+      assert(poset.edge?("B", "C"))
+      assert(poset.edge?("A", "C"))
+    end
+
+    def test_add_edge_not_reflexive
+      poset = Poset[String].new
+      poset.add_direct_edge("A", "B")
+
+      assert(poset.direct_edge?("A", "B"))
+      refute(poset.direct_edge?("B", "A"))
+
+      assert(poset.edge?("A", "B"))
+      refute(poset.edge?("B", "A"))
+    end
+
+    def test_update_edges
+      poset = Poset[String].new
+
+      poset.add_direct_edge("A", "B")
+      assert(poset.edge?("A", "B"))
+
+      poset.add_direct_edge("C", "D")
+      assert(poset.edge?("C", "D"))
+      refute(poset.edge?("A", "C"))
+      refute(poset.edge?("A", "D"))
+
+      poset.add_direct_edge("E", "C")
+      assert(poset.edge?("E", "C"))
+      assert(poset.edge?("E", "D"))
+
+      poset.add_direct_edge("B", "F")
+      assert(poset.direct_edge?("B", "F"))
+      assert(poset.edge?("B", "F"))
+      assert(poset.edge?("A", "F"))
+
+      poset.add_direct_edge("D", "F")
+      assert(poset.edge?("A", "F"))
+      assert(poset.edge?("B", "F"))
+      assert(poset.edge?("C", "F"))
+      assert(poset.edge?("D", "F"))
+      assert(poset.edge?("E", "F"))
+
+      poset.add_direct_edge("A", "C")
+      assert(poset.edge?("A", "F"))
+      assert(poset.edge?("B", "F"))
+      assert(poset.edge?("C", "F"))
+      assert(poset.edge?("D", "F"))
+      assert(poset.edge?("E", "F"))
+    end
+
+    def test_update_transitive_edges
+      poset = Poset[String].new
+
+      poset.add_direct_edge("A", "B")
+      poset.add_direct_edge("B", "C")
+      poset.add_direct_edge("D", "E")
+      poset.add_direct_edge("C", "D")
+
+      assert(poset.edge?("A", "B"))
+      assert(poset.edge?("A", "C"))
+      assert(poset.edge?("A", "D"))
+      assert(poset.edge?("A", "E"))
+    end
+
+    def test_get_element
+      poset = Poset[String].new
+
+      poset.add_direct_edge("A", "B")
+      poset.add_direct_edge("C", "D")
+      poset.add_direct_edge("E", "C")
+      poset.add_direct_edge("B", "F")
+      poset.add_direct_edge("D", "F")
+      poset.add_direct_edge("A", "C")
+
+      a = poset["A"]
+      assert_equal(["B", "C"], a.parents.sort)
+      assert_equal(["B", "C", "D", "F"], a.ancestors.sort)
+      assert_empty(a.children)
+      assert_empty(a.descendants)
+
+      b = poset["B"]
+      assert_equal(["F"], b.parents)
+      assert_equal(["F"], b.ancestors)
+      assert_equal(["A"], b.children)
+      assert_equal(["A"], b.descendants)
+
+      c = poset["C"]
+      assert_equal(["D"], c.parents)
+      assert_equal(["D", "F"], c.ancestors.sort)
+      assert_equal(["A", "E"], c.children.sort)
+      assert_equal(["A", "E"], c.descendants.sort)
+
+      d = poset["D"]
+      assert_equal(["F"], d.parents)
+      assert_equal(["F"], d.ancestors)
+      assert_equal(["C"], d.children)
+      assert_equal(["A", "C", "E"], d.descendants.sort)
+
+      e = poset["E"]
+      assert_equal(["C"], e.parents)
+      assert_equal(["C", "D", "F"], e.ancestors.sort)
+      assert_empty(e.children)
+      assert_empty(e.descendants)
+
+      f = poset["F"]
+      assert_empty(f.parents)
+      assert_empty(f.ancestors)
+      assert_equal(["B", "D"], f.children.sort)
+      assert_equal(["A", "B", "C", "D", "E"], f.descendants.sort)
+    end
+
+    def test_elements_comparison
+      poset = Poset[String].new
+
+      poset.add_direct_edge("A", "B")
+      poset.add_direct_edge("B", "C")
+      poset.add_direct_edge("D", "E")
+
+      a = poset["A"]
+      b = poset["B"]
+      c = poset["C"]
+      d = poset["D"]
+      e = poset["E"]
+
+      assert_equal(0, a <=> a) # rubocop:disable Lint/BinaryOperatorWithIdenticalOperands
+      assert_equal(-1, a <=> b)
+      assert_equal(1, b <=> a)
+      assert_equal(-1, a <=> c)
+      assert_equal(1, c <=> a)
+      assert_nil(a <=> e)
+      assert_nil(e <=> a)
+      assert_nil(a <=> d)
+      assert_nil(d <=> a)
+    end
+  end
+end