Started adding notifications and import jobs

db/migrate/025_add_notifications.rb

@@ -0,0 +1,15 @@
+Sequel.migration do
+  up do
+    create_table(:notifications) do
+      primary_key :id
+      String :message
+      String :url
+      TrueClass :seen
+      Time :created_at
+      Time :updated_at
+    end
+  end
+  down do
+    drop_table(:notifications)
+  end
+end

lib/coupler/base.rb

@@ -25,6 +25,7 @@ class Base < Sinatra::Base
     register Extensions::Jobs
     register Extensions::Transformers
     register Extensions::Imports
+    register Extensions::Notifications
     register Extensions::Exceptions
 
     helpers do

lib/coupler/extensions.rb

@@ -3,6 +3,8 @@ module Extensions
   end
 end
 
+require File.dirname(__FILE__) + "/extensions/exceptions"
+
 require File.dirname(__FILE__) + "/extensions/connections"
 require File.dirname(__FILE__) + "/extensions/projects"
 require File.dirname(__FILE__) + "/extensions/resources"
@@ -13,4 +15,4 @@ module Extensions
 require File.dirname(__FILE__) + "/extensions/jobs"
 require File.dirname(__FILE__) + "/extensions/transformers"
 require File.dirname(__FILE__) + "/extensions/imports"
-require File.dirname(__FILE__) + "/extensions/exceptions"
+require File.dirname(__FILE__) + "/extensions/notifications"

lib/coupler/extensions/imports.rb

@@ -9,19 +9,12 @@ def self.registered(app)
 
         app.post "/projects/:project_id/imports" do
           @import = Models::Import.new(params[:import].merge(:project_id => @project.id))
-
           if @import.save
-            @resource = Models::Resource.new(:import => @import)
-            if @resource.valid?
-              if @import.import!
-                @resource.save
-                redirect("/projects/#{@project.id}/resources/#{@resource.id}")
-              else
-                redirect("/projects/#{@project.id}/imports/#{@import.id}/edit")
-              end
-            end
+            Scheduler.instance.schedule_import_job(@import)
+            redirect("/projects/#{@project.id}")
+          else
+            erb(:'imports/new')
           end
-          erb(:'imports/new')
         end
 
         app.get "/projects/:project_id/imports/:id/edit" do

lib/coupler/extensions/notifications.rb

@@ -0,0 +1,14 @@
+module Coupler
+  module Extensions
+    module Notifications
+      include Models
+
+      def self.registered(app)
+        app.get "/notifications" do
+          @notifications = Notification.order(:created_at).all
+          erb :"notifications/index"
+        end
+      end
+    end
+  end
+end

lib/coupler/models.rb

@@ -16,6 +16,7 @@ module Models
     autoload :Scenario,       File.dirname(__FILE__) + "/models/scenario"
     autoload :Transformation, File.dirname(__FILE__) + "/models/transformation"
     autoload :Transformer,    File.dirname(__FILE__) + "/models/transformer"
+    autoload :Notification,   File.dirname(__FILE__) + "/models/notification"
   end
 end
 

lib/coupler/models/import.rb

@@ -49,7 +49,7 @@ def preview
         @preview
       end
 
-      def import!
+      def import!(&progress)
         project.local_database do |db|
           column_info = []
           column_names = []
@@ -78,7 +78,9 @@ def import!
           buffer = ImportBuffer.new(column_names, ds)
           skip = has_headers
           primary_key_index = field_names.index(primary_key_name)
-          FasterCSV.foreach(data.file.file) do |row|
+          io = File.open(data.file.file, 'rb')
+          csv = FasterCSV.new(io)
+          csv.each do |row|
             if skip
               # skip header if necessary
               skip = false
@@ -96,6 +98,10 @@ def import!
             self.has_duplicate_keys = true if num > 1
 
             buffer.add(row)
+
+            if block_given?
+              yield io.pos
+            end
           end
           buffer.flush
 

lib/coupler/models/job.rb

@@ -5,6 +5,7 @@ class Job < Sequel::Model
 
       many_to_one :resource
       many_to_one :scenario
+      many_to_one :import
 
       def percent_completed
         total > 0 ? completed * 100 / total : 0
@@ -25,6 +26,31 @@ def execute
           block = lambda do
             scenario.run!
           end
+
+        when 'import'
+          update(:status => 'running', :started_at => Time.now, :total => import.data.file.size)
+
+          new_status = 'failed'
+          begin
+            last = Time.now # don't slam the database
+            result = import.import! do |pos|
+              now = Time.now
+              if now - last >= 1
+                last = now
+                update(:completed => pos)
+              end
+            end
+            if !result
+              Notification.create({
+                :message => "Import finished, but with errors",
+                :url => "/projects/#{import.project_id}/imports/#{import.id}/edit"
+              })
+            end
+            new_status = 'done'
+          ensure
+            update(:status => new_status, :completed_at => Time.now)
+          end
+
         end
 
         begin

lib/coupler/models/notification.rb

@@ -0,0 +1,7 @@
+module Coupler
+  module Models
+    class Notification < Sequel::Model
+      include CommonModel
+    end
+  end
+end

lib/coupler/scheduler.rb

@@ -23,6 +23,14 @@ def schedule_run_scenario_job(scenario)
       })
     end
 
+    def schedule_import_job(import)
+      Models::Job.create({
+        :name => "import",
+        :import => import,
+        :status => "scheduled"
+      })
+    end
+
     def run_jobs
       @mutex.synchronize do
         count = Models::Job.filter(:status => 'running').count

test/functional/test_imports.rb

@@ -33,8 +33,10 @@ def setup
       find('label[for="resource-type-csv"]').click
       attach_file('data', fixture_path('people.csv'))
 
+      job_count = Job.count
       click_button('Begin Importing')
-      assert_match %r{^/projects/#{@project[:id]}/resources/\d+$}, page.current_path
+      assert_match %r{^/projects/#{@project[:id]}$}, page.current_path
+      assert_equal job_count + 1, Job.count
     end
 
     attribute(:javascript, true)

test/functional/test_notifications.rb

@@ -0,0 +1,20 @@
+require 'helper'
+
+module CouplerFunctionalTests
+  class TestNotifications < Coupler::Test::FunctionalTest
+
+    test "empty index" do
+      visit "/notifications"
+      assert_equal 200, page.status_code
+    end
+
+    test "index" do
+      n1 = Notification.create(:message => "Test!")
+      n2 = Notification.create(:message => "Another Test!", :url => "/projects")
+      n3 = Notification.create(:message => "Foo", :seen => true)
+      n4 = Notification.create(:message => "Bar", :url => "/connections", :seen => true)
+      visit "/notifications"
+      assert_equal 200, page.status_code
+    end
+  end
+end

test/integration/test_import.rb

@@ -5,7 +5,13 @@ class TestImport < Coupler::Test::IntegrationTest
   test "import!" do
     project = Project.create(:name => "foo")
     import = Import.create(:data => fixture_file_upload("people.csv"), :project => project)
-    import.import!
+
+    total = import.data.file.size
+    completed = 0
+    import.import! do |pos|
+      assert pos > completed
+      completed = pos
+    end
     assert_not_nil import.occurred_at
 
     project.local_database do |db|
@@ -22,6 +28,23 @@ class TestImport < Coupler::Test::IntegrationTest
     end
   end
 
+  test "import job" do
+    project = Project.create(:name => "foo")
+
+    tempfile = Tempfile.new('coupler-import')
+    tempfile.write("id,foo,bar\n1,2,3\n4,5,6\n7,8,9\n2,3,4\n5,6,7\n8,9,0\n")
+    tempfile.close
+    import = Import.create(:data => file_upload(tempfile.path), :project => project)
+
+    job = Job.create(:name => 'import', :import => import, :status => "scheduled")
+    job.execute
+
+    project.local_database do |db|
+      name = :"import_#{import.id}"
+      assert db.tables.include?(name)
+    end
+  end
+
   test "flags duplicate primary keys" do
     tempfile = Tempfile.new('coupler-import')
     tempfile.write("id,foo,bar\n1,abc,def\n2,ghi,jkl\n2,mno,pqr")

test/unit/models/test_job.rb

@@ -18,6 +18,9 @@ def setup
         super
         @resource = stub('resource', :pk => 456, :id => 456, :associations => {})
         @scenario = stub('scenario', :pk => 456, :id => 456, :associations => {})
+        @import = stub('import', :pk => 123, :id => 123, :associations => {}, :project_id => 1)
+        @notification = stub('notification')
+        Coupler::Models::Notification.stubs(:create).returns(@notification)
       end
 
       test "sequel model" do
@@ -33,6 +36,10 @@ def setup
         assert_respond_to Job.new, :scenario
       end
 
+      test "belongs to import" do
+        assert_respond_to Job.new, :import
+      end
+
       test "percent completed" do
         job = new_job(:name => 'transform', :resource => @resource, :total => 200, :completed => 54)
         assert_equal 27, job.percent_completed
@@ -112,6 +119,31 @@ def setup
         Timecop.freeze(now    ) { job_4 = new_job(:name => "run_scenario", :scenario => @scenario).save! }
         assert_equal [job_4, job_3, job_2], Job.recently_accessed
       end
+
+      test "import job" do
+        job = new_job(:name => 'import', :import => @import).save!
+
+        now = Time.now
+        seq = sequence("update")
+        @import.expects(:data).returns(mock(:file => mock(:size => 12345)))
+        job.expects(:update).with(:status => 'running', :total => 12345, :started_at => now).in_sequence(seq)
+        @import.expects(:import!).returns(true).in_sequence(seq)
+        job.expects(:update).with(:status => 'done', :completed_at => now).in_sequence(seq)
+        Timecop.freeze(now) { job.execute }
+      end
+
+      test "import job with more interaction needed" do
+        job = new_job(:name => 'import', :import => @import).save!
+
+        now = Time.now
+        seq = sequence("update")
+        @import.expects(:data).returns(mock(:file => mock(:size => 12345)))
+        job.expects(:update).with(:status => 'running', :total => 12345, :started_at => now).in_sequence(seq)
+        @import.expects(:import!).returns(false).in_sequence(seq)
+        Notification.expects(:create).with(:message => "Import finished, but with errors", :url => "/projects/1/imports/123/edit").returns(@notification)
+        job.expects(:update).with(:status => 'done', :completed_at => now).in_sequence(seq)
+        Timecop.freeze(now) { job.execute }
+      end
     end
   end
 end

test/unit/models/test_notification.rb

@@ -0,0 +1,17 @@
+require 'helper'
+
+module TestModels
+  class TestNotification < Coupler::Test::UnitTest
+    include Coupler::Models
+    Coupler::Models::Notification  # force load
+
+    test "sequel model" do
+      assert_equal ::Sequel::Model, Notification.superclass
+      assert_equal :notifications, Notification.table_name
+    end
+
+    test "common model" do
+      assert Notification.ancestors.include?(CommonModel)
+    end
+  end
+end

test/unit/test_scheduler.rb

@@ -24,6 +24,17 @@ class TestScheduler < Coupler::Test::UnitTest
       Scheduler.instance.schedule_run_scenario_job(scenario)
     end
 
+    test "schedule import job" do
+      import = mock('import')
+      Models::Job.expects(:create).with({
+        :name => "import",
+        :import => import,
+        :status => "scheduled"
+      })
+
+      Scheduler.instance.schedule_import_job(import)
+    end
+
     test "run_jobs executes first scheduled job" do
       running_dataset = mock('dataset')
       scheduled_dataset = mock('dataset')

webroot/views/jobs/list.erb

@@ -17,9 +17,10 @@
           <td><a href="/projects/<%= job.resource.project.id %>/resources/<%= job.resource.id %>"><%= job.resource.name %></a></td>
         <%- elsif job.scenario -%>
           <td><a href="/projects/<%= job.scenario.project.id %>/scenarios/<%= job.scenario.id %>"><%= job.scenario.name %></a></td>
-        <%- else -%>
-          <td>N/A</td>
-        <%- end -%>
+      <%- elsif job.import -%>
+        <td><%= job.import.name %> (Project: <a href="/projects/<%= job.import.project_id %>"><%= job.import.project.name %></a>)</td>
+      <%- else -%>
+        <td>N/A</td>
       <%- end -%>
       <td><%= timeago(job.created_at) %></td>
     </tr>

webroot/views/notifications/index.erb

@@ -0,0 +1,16 @@
+<%- @breadcrumbs = ["Notifications"] -%>
+<%- if @notifications.empty? -%>
+<p>There are currently no notifications.</p>
+<%- else -%>
+<ul class="notifications">
+  <%- @notifications.each do |notification| -%>
+    <li class="notification">
+      <% if notification.url %>
+        <a href="<%= notification.url %>"><%= notification.message %></a>
+      <% else %>
+        <%= notification.message %>
+      <% end %>
+    </li>
+  <%- end -%>
+</ul>
+<%- end -%>

webroot/views/sidebar.erb

@@ -97,6 +97,8 @@
             <a href="/jobs">Transform: <%= job.resource.name %></a>
           <%- elsif job.scenario -%>
             <a href="/jobs">Run: <%= job.scenario.name %></a>
+          <%- elsif job.import -%>
+            <a href="/jobs">Import: <%= job.import.name %></a>
           <%- end -%>
         </li>
       <%- end -%>