174 行
8.4 KiB
Ruby
174 行
8.4 KiB
Ruby
require 'rails_helper'
|
|
require 'tmpdir'
|
|
require 'csv'
|
|
|
|
RSpec.describe 'Counts export contracts' do
|
|
include ActiveSupport::Testing::TimeHelpers
|
|
let(:payload) do
|
|
{ 'resource' => 'posts', 'definition' => 'history', 'mode' => 'details',
|
|
'locale' => 'ja', 'execution_at' => '2024-03-01T00:00:00Z',
|
|
'columns' => 'title', 'headers' => '["Title"]', 'format' => 'json' }
|
|
end
|
|
|
|
around do |example|
|
|
Dir.mktmpdir('counts-export-spec') do |path|
|
|
@counts_export_dir = Pathname.new(path)
|
|
HistoryAccess.set(user: nil) { example.run }
|
|
end
|
|
end
|
|
|
|
before do
|
|
allow(CountsExportDownload).to receive(:directory)
|
|
.and_return(@counts_export_dir)
|
|
end
|
|
|
|
context 'serialization after the authorised query' do
|
|
let(:rows) { [{ title: "=formula\n\"quoted\"" }, { title: 'second' }] }
|
|
let(:page) do
|
|
{ count: 2, output_count: 2, visible_output_count: 2, page: 1, limit: 100,
|
|
versions: rows, execution_at: payload.fetch('execution_at') }
|
|
end
|
|
|
|
before do
|
|
query = CountsTableQuery.new(payload.slice(*CountsTableQuery::KEYS))
|
|
# SQL/authority are exercised in request specs; this unit boundary supplies their page.
|
|
allow(query).to receive(:each_page) do |before_page:, &block|
|
|
before_page.call
|
|
block.call(page)
|
|
end
|
|
allow(CountsTableQuery).to receive(:new).and_return(query)
|
|
allow(ApplicationRecord).to receive(:transaction).with(isolation: :repeatable_read).and_yield
|
|
end
|
|
|
|
it 'reports aggregation, actual output progress and regular-row metadata' do
|
|
progress = []
|
|
phases = []
|
|
body = CountsTableExport.new(payload).body(
|
|
progress: ->(done, total) { progress << [done, total] },
|
|
phase: ->(value) { phases << value }).to_a.join
|
|
result = JSON.parse(body)
|
|
expect(result['data']).to eq(rows.map(&:stringify_keys))
|
|
expect(result['metadata']).to include('logical_count_unit' => 'regular_result_rows',
|
|
'aggregate_records_in_logical_count' => false, 'output_count' => 2)
|
|
expect(phases).to eq(['aggregating', 'writing'])
|
|
expect(progress).to include([0, 2], [1, 2], [2, 2])
|
|
expect(progress.last).to eq([2, 2])
|
|
end
|
|
|
|
it 'preserves CSV multiline text and quotes while neutralising spreadsheet formulas' do
|
|
body = CountsTableExport.new(payload.merge('format' => 'csv')).body.to_a.join
|
|
expect(CSV.parse(body.delete_prefix("\uFEFF")))
|
|
.to eq([['Title'], ["'=formula\n\"quoted\""], ['second']])
|
|
end
|
|
|
|
it 'completes an isolated worker job with its real processed count and cancellable file' do
|
|
CountsExportJobs.pulse
|
|
token = CountsExportJobs.start(payload, 'owner', 'ja')
|
|
id = CountsExportJobs.authorised_id(token, 'owner')
|
|
state = CountsExportJobs.read(id).merge('status' => 'running', 'phase' => 'aggregating')
|
|
CountsExportJobs.write(state)
|
|
# Invoke the worker entrypoint directly; never fork or start an external worker in specs.
|
|
CountsExportJobs.perform(state)
|
|
ready = CountsExportJobs.status(token, 'owner')
|
|
expect(ready).to include('status' => 'ready', 'processed' => 2, 'total' => 2)
|
|
file = CountsExportDownload.find(ready.fetch('token'), 'owner')
|
|
expect(JSON.parse(File.read(file.fetch('path')))['data'].length).to eq(2)
|
|
CountsExportJobs.status(token, 'owner', cancel: true)
|
|
expect(CountsExportDownload.find(ready.fetch('token'), 'owner')).to be_nil
|
|
end
|
|
|
|
it 'marks a generation exception as failed without issuing a download token' do
|
|
CountsExportJobs.pulse
|
|
token = CountsExportJobs.start(payload, 'owner', 'ja')
|
|
id = CountsExportJobs.authorised_id(token, 'owner')
|
|
state = CountsExportJobs.read(id).merge('status' => 'running')
|
|
CountsExportJobs.write(state)
|
|
allow(CountsExportDownload).to receive(:prepare).and_raise(CountsExportDownload::TooLarge)
|
|
CountsExportJobs.perform(state)
|
|
expect(CountsExportJobs.status(token, 'owner')).to include('status' => 'failed')
|
|
expect(CountsExportJobs.status(token, 'owner')).not_to have_key('token')
|
|
end
|
|
end
|
|
|
|
it 'retains multilingual names and all inner rows in JSON with one selected display column' do
|
|
names = [{ language_code: 'ja', name: 'alpha', script_code: 'Jpan', aliases: [] },
|
|
{ language_code: 'ain', name: 'recorded', script_code: 'Kana',
|
|
aliases: [{ name: 'alias', script_code: 'Latn' }] }]
|
|
rows = [{ id: 1, name_ja: 'alpha', names: }, { id: 2, name_ja: 'beta', names: }]
|
|
projection = CountsExportProjection.new(payload.merge('columns' => 'tags',
|
|
'headers' => '["Tags"]', 'children' => 'tags.name_ja'),
|
|
resource: 'posts', summary: false, tags: false)
|
|
result = projection.project({ tags: rows })
|
|
expect(result['tags'].length).to eq(2)
|
|
expect(result['tags'].map { _1['name_ja'] }).to eq(['alpha', 'beta'])
|
|
expect(result['tags'].map { _1['names'] }).to eq([names, names])
|
|
end
|
|
|
|
it 'requires a live worker, enforces queue capacity and binds jobs to their owner' do
|
|
expect { CountsExportJobs.start(payload, 'owner', 'ja') }
|
|
.to raise_error(CountsExportJobs::WorkerUnavailable)
|
|
CountsExportJobs.pulse
|
|
token = CountsExportJobs.start(payload, 'owner', 'ja')
|
|
expect(CountsExportJobs.status(token, 'other')).to be_nil
|
|
expect(CountsExportJobs.status(token, 'owner')).to include(
|
|
'status' => 'queued', 'phase' => 'queued', 'processed' => 0, 'total' => nil)
|
|
(CountsExportDownload::MAX_CONCURRENT_EXPORTS - 1).times do
|
|
CountsExportJobs.start(payload, 'owner', 'ja')
|
|
end
|
|
expect { CountsExportJobs.start(payload, 'owner', 'ja') }
|
|
.to raise_error(CountsExportDownload::Busy)
|
|
expect(CountsExportJobs.status(token, 'owner', cancel: true)['status']).to eq('cancelled')
|
|
end
|
|
|
|
it 'fails stale queued/running jobs and preserves actual progress without a fake total' do
|
|
CountsExportJobs.pulse
|
|
token = CountsExportJobs.start(payload, 'owner', 'ja')
|
|
id = CountsExportJobs.authorised_id(token, 'owner')
|
|
state = CountsExportJobs.read(id)
|
|
state.merge!('status' => 'running', 'phase' => 'aggregating')
|
|
CountsExportJobs.write(state)
|
|
CountsExportJobs.checkpoint(id, 5, nil, phase: 'writing')
|
|
expect(CountsExportJobs.status(token, 'owner')).to include(
|
|
'phase' => 'writing', 'processed' => 5, 'total' => nil)
|
|
state = CountsExportJobs.read(id)
|
|
state['heartbeat_at'] = (Time.current - CountsExportJobs::HEARTBEAT_SECONDS - 1).to_f
|
|
CountsExportJobs.write(state)
|
|
expect(CountsExportJobs.status(token, 'owner')).to include(
|
|
'status' => 'failed', 'error' => 'worker_stalled', 'processed' => 5)
|
|
expect { CountsExportJobs.checkpoint(id, 6) }.to raise_error(CountsExportJobs::Cancelled)
|
|
end
|
|
|
|
it 'revokes an already generated download on a new discard and rejects wrong owners' do
|
|
exporter = instance_double(CountsExporter, resource: 'posts', format: 'json', body: ['{}'])
|
|
token = CountsExportDownload.prepare(exporter, 'owner')
|
|
expect(CountsExportDownload.find(token, 'owner')).not_to be_nil
|
|
expect(CountsExportDownload.find(token, 'other')).to be_nil
|
|
post = create(:post)
|
|
PostVersion.create!(post:, version_no: 1, event_type: 'discard', url: post.url,
|
|
title: 'private', tags: '', tags_json: [], parent_post_ids: '', created_at: Time.utc(2024))
|
|
expect(CountsExportDownload.find(token, 'owner')).to be_nil
|
|
end
|
|
|
|
it 'rejects oversized output and removes its partial file' do
|
|
stub_const('CountsExportDownload::MAX_FILE_BYTES', 4)
|
|
exporter = instance_double(CountsExporter, resource: 'posts', format: 'json', body: ['12345'])
|
|
expect { CountsExportDownload.prepare(exporter, 'owner') }
|
|
.to raise_error(CountsExportDownload::TooLarge)
|
|
expect(Dir.glob(CountsExportDownload.directory.join('counts-*'))).to be_empty
|
|
end
|
|
|
|
it 'expires queued work after worker loss and rejects expired download signatures' do
|
|
CountsExportJobs.pulse
|
|
token = CountsExportJobs.start(payload, 'owner', 'ja')
|
|
travel_to(Time.current + CountsExportJobs::HEARTBEAT_SECONDS + 1) do
|
|
expect(CountsExportJobs.status(token, 'owner')).to include(
|
|
'status' => 'failed', 'error' => 'worker_unavailable')
|
|
end
|
|
exporter = instance_double(CountsExporter, resource: 'posts', format: 'json', body: ['{}'])
|
|
download = CountsExportDownload.prepare(exporter, 'owner')
|
|
travel_to(Time.current + CountsExportDownload::EXPIRY + 1) do
|
|
expect(CountsExportDownload.find(download, 'owner')).to be_nil
|
|
end
|
|
end
|
|
end
|