test/functional/streamer.rb in mosql-0.3.2 vs test/functional/streamer.rb in mosql-0.4.0

@@ -51,18 +51,35 @@
             - 0
             - 0
     :columns:
       - _id: TEXT
       - var: INTEGER
+
+composite_key_test:
+  collection:
+    :meta:
+      :table: composite_table
+      :composite_key:
+        - store
+        - time
+    :columns:
+      - store:
+        :source: _id.s
+        :type: TEXT
+      - time:
+        :source: _id.t
+        :type: TIMESTAMP
+      - var: TEXT
 EOF
 
     before do
       @map = MoSQL::Schema.new(YAML.load(TEST_MAP))
       @adapter = MoSQL::SQLAdapter.new(@map, sql_test_uri)
 
       @sequel.drop_table?(:sqltable)
       @sequel.drop_table?(:sqltable2)
+      @sequel.drop_table?(:composite_table)
       @map.create_schema(@sequel)
 
       @streamer = build_streamer
     end
 
@@ -76,10 +93,26 @@
                             'o'  => { 'var' => 27 }
                           })
       assert_equal(27, sequel[:sqltable].where(:_id => o['_id'].to_s).select.first[:var])
     end
 
+    it 'applies ops performed via applyOps' do
+      o = { '_id' => BSON::ObjectId.new, 'var' => 17 }
+      @adapter.upsert_ns('mosql_test.collection', o)
+
+      op = { 'ns' => 'mosql_test.collection',
+             'op' => 'u',
+             'o2' => { '_id' => o['_id'] },
+             'o'  => { 'var' => 27 }
+           }
+      @streamer.handle_op({ 'op' => 'c',
+                            'ns' => 'mosql_test.$cmd',
+                            'o' => { 'applyOps' => [op] }
+                          })
+      assert_equal(27, sequel[:sqltable].where(:_id => o['_id'].to_s).select.first[:var])
+    end
+
     it 'handle "d" ops with BSON::ObjectIds' do
       o = { '_id' => BSON::ObjectId.new, 'var' => 17 }
       @adapter.upsert_ns('mosql_test.collection', o)
 
       @streamer.handle_op({ 'ns' => 'mosql_test.collection',
@@ -149,10 +182,43 @@
       record = inserted_records.first
       data[1][:_id] = data[1][:_id].to_s
       assert_equal(data[1], record)
     end
 
+    it 'handles "u" ops with a compsite key' do
+      date = Time.utc(2014, 7, 1)
+      o = {'_id' => {'s' => 'asdf', 't' => date}, 'var' => 'data'}
+      collection = mongo["composite_key_test"]["collection"]
+      collection.drop
+      collection.insert(o)
+
+      @streamer.options[:skip_tail] = true
+      @streamer.initial_import
+
+      collection.update({ '_id' => { 's' => 'asdf', 't' => date}}, { '$set' => { 'var' => 'new_data'}})
+      @streamer.handle_op({'ns' => 'composite_key_test.collection',
+                           'op' => 'u',
+                           'o2' => { '_id' => { 's' => 'asdf', 't' => date}},
+                           'o'  => { '$set' => { 'var' => 'new_data'}}
+                           })
+
+      assert_equal(0, @sequel[:composite_table].where(:var => "data").count)
+      assert_equal(1, @sequel[:composite_table].where(:var => "new_data").count)
+    end
+
+    it 'handles composite keys' do
+      o = {'_id' => {'s' => 'asdf', 't' => Time.new}, 'var' => 'data'}
+      collection = mongo["composite_key_test"]["collection"]
+      collection.drop
+      collection.insert(o)
+
+      @streamer.options[:skip_tail] = true
+      @streamer.initial_import
+
+      assert_equal(1, @sequel[:composite_table].count)
+    end
+
     describe '.bulk_upsert' do
       it 'inserts multiple rows' do
         objs = [
                 { '_id' => BSON::ObjectId.new, 'var' => 0 },
                 { '_id' => BSON::ObjectId.new, 'var' => 1, 'arry' => [1, 2] },
@@ -248,8 +314,60 @@
       @streamer.options[:skip_tail] = true
       @streamer.initial_import
 
       sqlobjs = @sequel[:sqltable].select.to_a
       assert_equal(ids.map(&:to_s).sort, sqlobjs.map { |o| o[:_id] }.sort)
+    end
+  end
+  describe 'timestamps' do
+  TIMESTAMP_MAP = <<EOF
+---
+db:
+  has_timestamp:
+    :meta:
+      :table: has_timestamp
+    :columns:
+      - _id: TEXT
+      - ts: timestamp
+EOF
+
+    before do
+      @map = MoSQL::Schema.new(YAML.load(TIMESTAMP_MAP))
+      @adapter = MoSQL::SQLAdapter.new(@map, sql_test_uri)
+
+      mongo['db']['has_timestamp'].drop
+      @sequel.drop_table?(:has_timestamp)
+      @map.create_schema(@sequel)
+
+      @streamer = build_streamer
+    end
+
+    it 'preserves milliseconds on import' do
+      ts = Time.utc(2014, 8, 7, 6, 54, 32, 123000)
+      mongo['db']['has_timestamp'].insert({ts: ts})
+      @streamer.options[:skip_tail] = true
+      @streamer.initial_import
+
+      row = @sequel[:has_timestamp].select.to_a
+      assert_equal(1, row.length)
+      assert_equal(ts.to_i, row.first[:ts].to_i)
+      assert_equal(ts.tv_usec, row.first[:ts].tv_usec)
+    end
+
+    it 'preserves milliseconds on tailing' do
+      ts = Time.utc(2006,01,02, 15,04,05,678000)
+      id = mongo['db']['has_timestamp'].insert({ts: ts})
+      @streamer.handle_op(
+        {
+          "ts" => {"t" => 1408647630, "i" => 4},
+          "h"  => -965650193548512059,
+          "v"  => 2,
+          "op" => "i",
+          "ns" => "db.has_timestamp",
+          "o"  => mongo['db']['has_timestamp'].find_one({_id: id})
+        })
+      got = @sequel[:has_timestamp].where(:_id => id.to_s).select.first[:ts]
+      assert_equal(ts.to_i, got.to_i)
+      assert_equal(ts.tv_usec, got.tv_usec)
     end
   end
 end