Commit 9c5a922e by Calen Pennington

Create tables for all known scopes, and add tests of the LmsKeyValueStore

parent 6e933ae7
# -*- coding: utf-8 -*-
import datetime
from south.db import db
from south.v2 import SchemaMigration
from django.db import models
class Migration(SchemaMigration):
def forwards(self, orm):
# Adding model 'XModuleStudentInfoField'
db.create_table('courseware_xmodulestudentinfofield', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('field_name', self.gf('django.db.models.fields.CharField')(max_length=64, db_index=True)),
('value', self.gf('django.db.models.fields.TextField')(null=True, blank=True)),
('student', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.User'])),
('created', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, db_index=True, blank=True)),
('modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, db_index=True, blank=True)),
))
db.send_create_signal('courseware', ['XModuleStudentInfoField'])
# Adding unique constraint on 'XModuleStudentInfoField', fields ['student', 'field_name']
db.create_unique('courseware_xmodulestudentinfofield', ['student_id', 'field_name'])
# Adding model 'XModuleContentField'
db.create_table('courseware_xmodulecontentfield', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('field_name', self.gf('django.db.models.fields.CharField')(max_length=64, db_index=True)),
('definition_id', self.gf('django.db.models.fields.CharField')(max_length=255, db_index=True)),
('value', self.gf('django.db.models.fields.TextField')(null=True, blank=True)),
('created', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, db_index=True, blank=True)),
('modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, db_index=True, blank=True)),
))
db.send_create_signal('courseware', ['XModuleContentField'])
# Adding unique constraint on 'XModuleContentField', fields ['definition_id', 'field_name']
db.create_unique('courseware_xmodulecontentfield', ['definition_id', 'field_name'])
# Adding model 'XModuleSettingsField'
db.create_table('courseware_xmodulesettingsfield', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('field_name', self.gf('django.db.models.fields.CharField')(max_length=64, db_index=True)),
('usage_id', self.gf('django.db.models.fields.CharField')(max_length=255, db_index=True)),
('value', self.gf('django.db.models.fields.TextField')(null=True, blank=True)),
('created', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, db_index=True, blank=True)),
('modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, db_index=True, blank=True)),
))
db.send_create_signal('courseware', ['XModuleSettingsField'])
# Adding unique constraint on 'XModuleSettingsField', fields ['usage_id', 'field_name']
db.create_unique('courseware_xmodulesettingsfield', ['usage_id', 'field_name'])
# Adding model 'XModuleStudentPrefsField'
db.create_table('courseware_xmodulestudentprefsfield', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('field_name', self.gf('django.db.models.fields.CharField')(max_length=64, db_index=True)),
('module_type', self.gf('django.db.models.fields.CharField')(max_length=64, db_index=True)),
('value', self.gf('django.db.models.fields.TextField')(null=True, blank=True)),
('student', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.User'])),
('created', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, db_index=True, blank=True)),
('modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, db_index=True, blank=True)),
))
db.send_create_signal('courseware', ['XModuleStudentPrefsField'])
# Adding unique constraint on 'XModuleStudentPrefsField', fields ['student', 'module_type', 'field_name']
db.create_unique('courseware_xmodulestudentprefsfield', ['student_id', 'module_type', 'field_name'])
def backwards(self, orm):
# Removing unique constraint on 'XModuleStudentPrefsField', fields ['student', 'module_type', 'field_name']
db.delete_unique('courseware_xmodulestudentprefsfield', ['student_id', 'module_type', 'field_name'])
# Removing unique constraint on 'XModuleSettingsField', fields ['usage_id', 'field_name']
db.delete_unique('courseware_xmodulesettingsfield', ['usage_id', 'field_name'])
# Removing unique constraint on 'XModuleContentField', fields ['definition_id', 'field_name']
db.delete_unique('courseware_xmodulecontentfield', ['definition_id', 'field_name'])
# Removing unique constraint on 'XModuleStudentInfoField', fields ['student', 'field_name']
db.delete_unique('courseware_xmodulestudentinfofield', ['student_id', 'field_name'])
# Deleting model 'XModuleStudentInfoField'
db.delete_table('courseware_xmodulestudentinfofield')
# Deleting model 'XModuleContentField'
db.delete_table('courseware_xmodulecontentfield')
# Deleting model 'XModuleSettingsField'
db.delete_table('courseware_xmodulesettingsfield')
# Deleting model 'XModuleStudentPrefsField'
db.delete_table('courseware_xmodulestudentprefsfield')
models = {
'auth.group': {
'Meta': {'object_name': 'Group'},
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
},
'auth.permission': {
'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
},
'auth.user': {
'Meta': {'object_name': 'User'},
'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
},
'contenttypes.contenttype': {
'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
},
'courseware.studentmodule': {
'Meta': {'unique_together': "(('student', 'module_state_key', 'course_id'),)", 'object_name': 'StudentModule'},
'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'done': ('django.db.models.fields.CharField', [], {'default': "'na'", 'max_length': '8', 'db_index': 'True'}),
'grade': ('django.db.models.fields.FloatField', [], {'db_index': 'True', 'null': 'True', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'max_grade': ('django.db.models.fields.FloatField', [], {'null': 'True', 'blank': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'module_state_key': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_column': "'module_id'", 'db_index': 'True'}),
'module_type': ('django.db.models.fields.CharField', [], {'default': "'problem'", 'max_length': '32', 'db_index': 'True'}),
'state': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
'student': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"})
},
'courseware.xmodulecontentfield': {
'Meta': {'unique_together': "(('definition_id', 'field_name'),)", 'object_name': 'XModuleContentField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'definition_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'value': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'})
},
'courseware.xmodulesettingsfield': {
'Meta': {'unique_together': "(('usage_id', 'field_name'),)", 'object_name': 'XModuleSettingsField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'usage_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'value': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'})
},
'courseware.xmodulestudentinfofield': {
'Meta': {'unique_together': "(('student', 'field_name'),)", 'object_name': 'XModuleStudentInfoField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'student': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
'value': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'})
},
'courseware.xmodulestudentprefsfield': {
'Meta': {'unique_together': "(('student', 'module_type', 'field_name'),)", 'object_name': 'XModuleStudentPrefsField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'module_type': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'student': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
'value': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'})
}
}
complete_apps = ['courseware']
\ No newline at end of file
# -*- coding: utf-8 -*-
import datetime
from south.db import db
from south.v2 import SchemaMigration
from django.db import models
class Migration(SchemaMigration):
def forwards(self, orm):
# Changing field 'XModuleContentField.value'
db.alter_column('courseware_xmodulecontentfield', 'value', self.gf('django.db.models.fields.TextField')())
# Changing field 'XModuleStudentInfoField.value'
db.alter_column('courseware_xmodulestudentinfofield', 'value', self.gf('django.db.models.fields.TextField')())
# Changing field 'XModuleSettingsField.value'
db.alter_column('courseware_xmodulesettingsfield', 'value', self.gf('django.db.models.fields.TextField')())
# Changing field 'XModuleStudentPrefsField.value'
db.alter_column('courseware_xmodulestudentprefsfield', 'value', self.gf('django.db.models.fields.TextField')())
def backwards(self, orm):
# Changing field 'XModuleContentField.value'
db.alter_column('courseware_xmodulecontentfield', 'value', self.gf('django.db.models.fields.TextField')(null=True))
# Changing field 'XModuleStudentInfoField.value'
db.alter_column('courseware_xmodulestudentinfofield', 'value', self.gf('django.db.models.fields.TextField')(null=True))
# Changing field 'XModuleSettingsField.value'
db.alter_column('courseware_xmodulesettingsfield', 'value', self.gf('django.db.models.fields.TextField')(null=True))
# Changing field 'XModuleStudentPrefsField.value'
db.alter_column('courseware_xmodulestudentprefsfield', 'value', self.gf('django.db.models.fields.TextField')(null=True))
models = {
'auth.group': {
'Meta': {'object_name': 'Group'},
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
},
'auth.permission': {
'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
},
'auth.user': {
'Meta': {'object_name': 'User'},
'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
},
'contenttypes.contenttype': {
'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
},
'courseware.studentmodule': {
'Meta': {'unique_together': "(('student', 'module_state_key', 'course_id'),)", 'object_name': 'StudentModule'},
'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'done': ('django.db.models.fields.CharField', [], {'default': "'na'", 'max_length': '8', 'db_index': 'True'}),
'grade': ('django.db.models.fields.FloatField', [], {'db_index': 'True', 'null': 'True', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'max_grade': ('django.db.models.fields.FloatField', [], {'null': 'True', 'blank': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'module_state_key': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_column': "'module_id'", 'db_index': 'True'}),
'module_type': ('django.db.models.fields.CharField', [], {'default': "'problem'", 'max_length': '32', 'db_index': 'True'}),
'state': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
'student': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"})
},
'courseware.xmodulecontentfield': {
'Meta': {'unique_together': "(('definition_id', 'field_name'),)", 'object_name': 'XModuleContentField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'definition_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'value': ('django.db.models.fields.TextField', [], {'default': "'null'"})
},
'courseware.xmodulesettingsfield': {
'Meta': {'unique_together': "(('usage_id', 'field_name'),)", 'object_name': 'XModuleSettingsField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'usage_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'value': ('django.db.models.fields.TextField', [], {'default': "'null'"})
},
'courseware.xmodulestudentinfofield': {
'Meta': {'unique_together': "(('student', 'field_name'),)", 'object_name': 'XModuleStudentInfoField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'student': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
'value': ('django.db.models.fields.TextField', [], {'default': "'null'"})
},
'courseware.xmodulestudentprefsfield': {
'Meta': {'unique_together': "(('student', 'module_type', 'field_name'),)", 'object_name': 'XModuleStudentPrefsField'},
'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'db_index': 'True', 'blank': 'True'}),
'field_name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
'module_type': ('django.db.models.fields.CharField', [], {'max_length': '64', 'db_index': 'True'}),
'student': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
'value': ('django.db.models.fields.TextField', [], {'default': "'null'"})
}
}
complete_apps = ['courseware']
\ No newline at end of file
import json
from collections import namedtuple
from .models import (
StudentModule,
XModuleContentField,
XModuleSettingsField,
XModuleStudentPrefsField,
XModuleStudentInfoField
)
from xmodule.runtime import DbModel, KeyValueStore
from xmodule.model import Scope
class InvalidScopeError(Exception):
pass
class InvalidWriteError(Exception):
pass
class LmsKeyValueStore(KeyValueStore):
"""
This KeyValueStore will read data from descriptor_model_data if it exists,
but will not overwrite any keys set in descriptor_model_data. Attempts to do so will
raise an InvalidWriteError.
If the scope to write to is not one of the 5 named scopes:
Scope.content
Scope.settings
Scope.student_state
Scope.student_preferences
Scope.student_info
then an InvalidScopeError will be raised.
Data for Scope.student_state is stored as StudentModule objects via the django orm.
Data for the other scopes is stored in individual objects that are named for the
scope involved and have the field name as a key
If the key isn't found in the expected table during a read or a delete, then a KeyError will be raised
"""
def __init__(self, course_id, user, descriptor_model_data, student_module_cache):
self._course_id = course_id
self._user = user
self._descriptor_model_data = descriptor_model_data
self._student_module_cache = student_module_cache
def _student_module(self, key):
student_module = self._student_module_cache.lookup(
self._course_id, key.module_scope_id.category, key.module_scope_id.url()
)
return student_module
def _field_object(self, key):
if key.scope == Scope.content:
return XModuleContentField, {'field_name': key.field_name, 'definition_id': key.module_scope_id}
elif key.scope == Scope.settings:
return XModuleSettingsField, {
'field_name': key.field_name,
'usage_id': '%s-%s' % (self._course_id, key.module_scope_id)
}
elif key.scope == Scope.student_preferences:
return XModuleStudentPrefsField, {'field_name': key.field_name, 'student': self._user, 'module_type': key.module_scope_id}
elif key.scope == Scope.student_info:
return XModuleStudentInfoField, {'field_name': key.field_name, 'student': self._user}
raise InvalidScopeError(key.scope)
def get(self, key):
if key.field_name in self._descriptor_model_data:
return self._descriptor_model_data[key.field_name]
if key.scope == Scope.student_state:
student_module = self._student_module(key)
if student_module is None:
raise KeyError(key.field_name)
return json.loads(student_module.state)[key.field_name]
scope_field_cls, search_kwargs = self._field_object(key)
try:
return json.loads(scope_field_cls.objects.get(**search_kwargs).value)
except scope_field_cls.DoesNotExist:
raise KeyError(key.field_name)
def set(self, key, value):
if key.field_name in self._descriptor_model_data:
raise InvalidWriteError("Not allowed to overwrite descriptor model data", key.field_name)
if key.scope == Scope.student_state:
student_module = self._student_module(key)
if student_module is None:
student_module = StudentModule(
course_id=self._course_id,
student=self._user,
module_type=key.module_scope_id.category,
module_state_key=key.module_scope_id.url(),
state=json.dumps({})
)
self._student_module_cache.append(student_module)
state = json.loads(student_module.state)
state[key.field_name] = value
student_module.state = json.dumps(state)
student_module.save()
return
scope_field_cls, search_kwargs = self._field_object(key)
json_value = json.dumps(value)
field, created = scope_field_cls.objects.select_for_update().get_or_create(
defaults={'value': json_value},
**search_kwargs
)
if not created:
field.value = json_value
field.save()
def delete(self, key):
if key.field_name in self._descriptor_model_data:
raise InvalidWriteError("Not allowed to deleted descriptor model data", key.field_name)
if key.scope == Scope.student_state:
student_module = self._student_module(key)
if student_module is None:
raise KeyError(key.field_name)
state = json.loads(student_module.state)
del state[key.field_name]
student_module.state = json.dumps(state)
student_module.save()
return
scope_field_cls, search_kwargs = self._field_object(key)
print scope_field_cls, search_kwargs
query = scope_field_cls.objects.filter(**search_kwargs)
if not query.exists():
raise KeyError(key.field_name)
query.delete()
LmsUsage = namedtuple('LmsUsage', 'id, def_id')
......@@ -72,6 +72,132 @@ class StudentModule(models.Model):
def __unicode__(self):
return unicode(repr(self))
class XModuleContentField(models.Model):
"""
Stores data set in the Scope.content scope by an xmodule field
"""
class Meta:
unique_together = (('definition_id', 'field_name'),)
# The name of the field
field_name = models.CharField(max_length=64, db_index=True)
# The definition id for the module
definition_id = models.CharField(max_length=255, db_index=True)
# The value of the field. Defaults to None dumped as json
value = models.TextField(default='null')
created = models.DateTimeField(auto_now_add=True, db_index=True)
modified = models.DateTimeField(auto_now=True, db_index=True)
def __repr__(self):
return 'XModuleContentField<%r>' % ({
'field_name': self.field_name,
'definition_id': self.definition_id,
'value': self.value,
},)
def __unicode__(self):
return unicode(repr(self))
class XModuleSettingsField(models.Model):
"""
Stores data set in the Scope.settings scope by an xmodule field
"""
class Meta:
unique_together = (('usage_id', 'field_name'),)
# The name of the field
field_name = models.CharField(max_length=64, db_index=True)
# The usage id for the module
usage_id = models.CharField(max_length=255, db_index=True)
# The value of the field. Defaults to None, dumped as json
value = models.TextField(default='null')
created = models.DateTimeField(auto_now_add=True, db_index=True)
modified = models.DateTimeField(auto_now=True, db_index=True)
def __repr__(self):
return 'XModuleSettingsField<%r>' % ({
'field_name': self.field_name,
'usage_id': self.usage_id,
'value': self.value,
},)
def __unicode__(self):
return unicode(repr(self))
class XModuleStudentPrefsField(models.Model):
"""
Stores data set in the Scope.student_preferences scope by an xmodule field
"""
class Meta:
unique_together = (('student', 'module_type', 'field_name'),)
# The name of the field
field_name = models.CharField(max_length=64, db_index=True)
# The type of the module for these preferences
module_type = models.CharField(max_length=64, db_index=True)
# The value of the field. Defaults to None dumped as json
value = models.TextField(default='null')
student = models.ForeignKey(User, db_index=True)
created = models.DateTimeField(auto_now_add=True, db_index=True)
modified = models.DateTimeField(auto_now=True, db_index=True)
def __repr__(self):
return 'XModuleStudentPrefsField<%r>' % ({
'field_name': self.field_name,
'module_type': self.module_type,
'student': self.student.username,
'value': self.value,
},)
def __unicode__(self):
return unicode(repr(self))
class XModuleStudentInfoField(models.Model):
"""
Stores data set in the Scope.student_preferences scope by an xmodule field
"""
class Meta:
unique_together = (('student', 'field_name'),)
# The name of the field
field_name = models.CharField(max_length=64, db_index=True)
# The value of the field. Defaults to None dumped as json
value = models.TextField(default='null')
student = models.ForeignKey(User, db_index=True)
created = models.DateTimeField(auto_now_add=True, db_index=True)
modified = models.DateTimeField(auto_now=True, db_index=True)
def __repr__(self):
return 'XModuleStudentInfoField<%r>' % ({
'field_name': self.field_name,
'student': self.student.username,
'value': self.value,
},)
def __unicode__(self):
return unicode(repr(self))
# TODO (cpennington): Remove these once the LMS switches to using XModuleDescriptors
......
......@@ -11,8 +11,6 @@ from django.http import Http404
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from collections import namedtuple
from requests.auth import HTTPBasicAuth
from capa.xqueue_interface import XQueueInterface
......@@ -28,9 +26,9 @@ from xmodule.modulestore import Location
from xmodule.modulestore.django import modulestore
from xmodule.x_module import ModuleSystem
from xmodule.error_module import ErrorDescriptor, NonStaffErrorDescriptor
from xmodule.runtime import DbModel, KeyValueStore
from xmodule.model import Scope
from xmodule.runtime import DbModel
from xmodule_modifiers import replace_course_urls, replace_static_urls, add_histogram, wrap_xmodule
from .model_data import LmsKeyValueStore, LmsUsage
from xmodule.modulestore.exceptions import ItemNotFoundError
from statsd import statsd
......@@ -149,70 +147,6 @@ def get_module(user, request, location, student_module_cache, course_id, positio
return None
class LmsKeyValueStore(KeyValueStore):
def __init__(self, course_id, user, descriptor_model_data, student_module_cache):
self._course_id = course_id
self._user = user
self._descriptor_model_data = descriptor_model_data
self._student_module_cache = student_module_cache
def _student_module(self, key):
student_module = self._student_module_cache.lookup(
self._course_id, key.module_scope_id.category, key.module_scope_id.url()
)
return student_module
def get(self, key):
if not key.scope.student:
return self._descriptor_model_data[key.field_name]
if key.scope == Scope.student_state:
student_module = self._student_module(key)
if student_module is None:
raise KeyError(key.field_name)
return json.loads(student_module.state)[key.field_name]
def set(self, key, value):
if not key.scope.student:
self._descriptor_model_data[key.field_name] = value
if key.scope == Scope.student_state:
student_module = self._student_module(key)
if student_module is None:
student_module = StudentModule(
course_id=self._course_id,
student=self._user,
module_type=key.module_scope_id.category,
module_state_key=key.module_scope_id.url(),
state=json.dumps({})
)
self._student_module_cache.append(student_module)
state = json.loads(student_module.state)
state[key.field_name] = value
student_module.state = json.dumps(state)
student_module.save()
def delete(self, key):
if not key.scope.student:
del self._descriptor_model_data[key.field_name]
if key.scope == Scope.student_state:
student_module = self._student_module(key)
if student_module is None:
raise KeyError(key.field_name)
state = json.loads(student_module.state)
del state[key.field_name]
student_module.state = json.dumps(state)
student_module.save()
LmsUsage = namedtuple('LmsUsage', 'id, def_id')
def _get_module(user, request, location, student_module_cache, course_id, position=None, wrap_xmodule_display=True):
"""
Actually implement get_module. See docstring there for details.
......
import factory
import json
from mock import Mock
from django.contrib.auth.models import User
from functools import partial
from courseware.model_data import LmsKeyValueStore, InvalidWriteError, InvalidScopeError
from courseware.models import StudentModule, XModuleContentField, XModuleSettingsField, XModuleStudentInfoField, XModuleStudentPrefsField, StudentModuleCache
from xmodule.model import Scope, ModuleScope
from xmodule.modulestore import Location
from django.test import TestCase
def mock_descriptor():
descriptor = Mock()
descriptor.stores_state = True
descriptor.location = location('def_id')
return descriptor
location = partial(Location, 'i4x', 'edX', 'test_course', 'problem')
course_id = 'edX/test_course/test'
content_key = partial(LmsKeyValueStore.Key, Scope.content, None, location('def_id'))
settings_key = partial(LmsKeyValueStore.Key, Scope.settings, None, location('def_id'))
student_state_key = partial(LmsKeyValueStore.Key, Scope.student_state, 'user', location('def_id'))
student_prefs_key = partial(LmsKeyValueStore.Key, Scope.student_preferences, 'user', 'problem')
student_info_key = partial(LmsKeyValueStore.Key, Scope.student_info, 'user', None)
class UserFactory(factory.Factory):
FACTORY_FOR = User
username = 'user'
class StudentModuleFactory(factory.Factory):
FACTORY_FOR = StudentModule
module_type = 'problem'
module_state_key = location('def_id').url()
student = factory.SubFactory(UserFactory)
course_id = course_id
state = None
class ContentFactory(factory.Factory):
FACTORY_FOR = XModuleContentField
field_name = 'content_field'
value = json.dumps('content_value')
definition_id = location('def_id').url()
class SettingsFactory(factory.Factory):
FACTORY_FOR = XModuleSettingsField
field_name = 'settings_field'
value = json.dumps('settings_value')
usage_id = '%s-%s' % (course_id, location('def_id').url())
class StudentPrefsFactory(factory.Factory):
FACTORY_FOR = XModuleStudentPrefsField
field_name = 'student_pref_field'
value = json.dumps('student_pref_value')
student = factory.SubFactory(UserFactory)
module_type = 'problem'
class StudentInfoFactory(factory.Factory):
FACTORY_FOR = XModuleStudentInfoField
field_name = 'student_info_field'
value = json.dumps('student_info_value')
student = factory.SubFactory(UserFactory)
class TestDescriptorFallback(TestCase):
def setUp(self):
self.desc_md = {
'field_a': 'content',
'field_b': 'settings',
}
self.kvs = LmsKeyValueStore(course_id, UserFactory.build(), self.desc_md, None)
def test_get_from_descriptor(self):
self.assertEquals('content', self.kvs.get(content_key('field_a')))
self.assertEquals('settings', self.kvs.get(settings_key('field_b')))
def test_write_to_descriptor(self):
self.assertRaises(InvalidWriteError, self.kvs.set, content_key('field_a'), 'foo')
self.assertEquals('content', self.desc_md['field_a'])
self.assertRaises(InvalidWriteError, self.kvs.set, settings_key('field_b'), 'foo')
self.assertEquals('settings', self.desc_md['field_b'])
self.assertRaises(InvalidWriteError, self.kvs.delete, content_key('field_a'))
self.assertEquals('content', self.desc_md['field_a'])
self.assertRaises(InvalidWriteError, self.kvs.delete, settings_key('field_b'))
self.assertEquals('settings', self.desc_md['field_b'])
class TestStudentStateFields(TestCase):
pass
class TestInvalidScopes(TestCase):
def setUp(self):
self.desc_md = {}
self.kvs = LmsKeyValueStore(course_id, UserFactory.build(), self.desc_md, None)
def test_invalid_scopes(self):
for scope in (Scope(student=True, module=ModuleScope.DEFINITION),
Scope(student=False, module=ModuleScope.TYPE),
Scope(student=False, module=ModuleScope.ALL)):
self.assertRaises(InvalidScopeError, self.kvs.get, LmsKeyValueStore.Key(scope, None, None, 'field'))
self.assertRaises(InvalidScopeError, self.kvs.set, LmsKeyValueStore.Key(scope, None, None, 'field'), 'value')
self.assertRaises(InvalidScopeError, self.kvs.delete, LmsKeyValueStore.Key(scope, None, None, 'field'))
class TestStudentModuleStorage(TestCase):
def setUp(self):
student_module = StudentModuleFactory.create(state=json.dumps({'a_field': 'a_value'}))
self.user = student_module.student
self.desc_md = {}
self.smc = StudentModuleCache(course_id, self.user, [mock_descriptor()])
self.kvs = LmsKeyValueStore(course_id, self.user, self.desc_md, self.smc)
def test_get_existing_field(self):
"Test that getting an existing field in an existing StudentModule works"
self.assertEquals('a_value', self.kvs.get(student_state_key('a_field')))
def test_get_missing_field(self):
"Test that getting a missing field from an existing StudentModule raises a KeyError"
self.assertRaises(KeyError, self.kvs.get, student_state_key('not_a_field'))
def test_set_existing_field(self):
"Test that setting an existing student_state field changes the value"
self.kvs.set(student_state_key('a_field'), 'new_value')
self.assertEquals(1, StudentModule.objects.all().count())
self.assertEquals({'a_field': 'new_value'}, json.loads(StudentModule.objects.all()[0].state))
def test_set_missing_field(self):
"Test that setting a new student_state field changes the value"
self.kvs.set(student_state_key('not_a_field'), 'new_value')
self.assertEquals(1, StudentModule.objects.all().count())
self.assertEquals({'a_field': 'a_value', 'not_a_field': 'new_value'}, json.loads(StudentModule.objects.all()[0].state))
def test_delete_existing_field(self):
"Test that deleting an existing field removes it from the StudentModule"
self.kvs.delete(student_state_key('a_field'))
self.assertEquals(1, StudentModule.objects.all().count())
self.assertEquals({}, json.loads(StudentModule.objects.all()[0].state))
def test_delete_missing_field(self):
"Test that deleting a missing field from an existing StudentModule raises a KeyError"
self.assertRaises(KeyError, self.kvs.delete, student_state_key('not_a_field'))
self.assertEquals(1, StudentModule.objects.all().count())
self.assertEquals({'a_field': 'a_value'}, json.loads(StudentModule.objects.all()[0].state))
class TestMissingStudentModule(TestCase):
def setUp(self):
self.user = UserFactory.create()
self.desc_md = {}
self.smc = StudentModuleCache(course_id, self.user, [mock_descriptor()])
self.kvs = LmsKeyValueStore(course_id, self.user, self.desc_md, self.smc)
def test_get_field_from_missing_student_module(self):
"Test that getting a field from a missing StudentModule raises a KeyError"
self.assertRaises(KeyError, self.kvs.get, student_state_key('a_field'))
def test_set_field_in_missing_student_module(self):
"Test that setting a field in a missing StudentModule creates the student module"
self.assertEquals(0, len(self.smc.cache))
self.assertEquals(0, StudentModule.objects.all().count())
self.kvs.set(student_state_key('a_field'), 'a_value')
self.assertEquals(1, len(self.smc.cache))
self.assertEquals(1, StudentModule.objects.all().count())
student_module = StudentModule.objects.all()[0]
self.assertEquals({'a_field': 'a_value'}, json.loads(student_module.state))
self.assertEquals(self.user, student_module.student)
self.assertEquals(location('def_id').url(), student_module.module_state_key)
self.assertEquals(course_id, student_module.course_id)
def test_delete_field_from_missing_student_module(self):
"Test that deleting a field from a missing StudentModule raises a KeyError"
self.assertRaises(KeyError, self.kvs.delete, student_state_key('a_field'))
class TestSettingsStorage(TestCase):
def setUp(self):
settings = SettingsFactory.create()
self.user = UserFactory.create()
self.desc_md = {}
self.smc = StudentModuleCache(course_id, self.user, [])
self.kvs = LmsKeyValueStore(course_id, self.user, self.desc_md, self.smc)
def test_get_existing_field(self):
"Test that getting an existing field in an existing SettingsField works"
self.assertEquals('settings_value', self.kvs.get(settings_key('settings_field')))
def test_get_missing_field(self):
"Test that getting a missing field from an existing SettingsField raises a KeyError"
self.assertRaises(KeyError, self.kvs.get, settings_key('not_settings_field'))
def test_set_existing_field(self):
"Test that setting an existing field changes the value"
self.kvs.set(settings_key('settings_field'), 'new_value')
self.assertEquals(1, XModuleSettingsField.objects.all().count())
self.assertEquals('new_value', json.loads(XModuleSettingsField.objects.all()[0].value))
def test_set_missing_field(self):
"Test that setting a new field changes the value"
self.kvs.set(settings_key('not_settings_field'), 'new_value')
self.assertEquals(2, XModuleSettingsField.objects.all().count())
self.assertEquals('settings_value', json.loads(XModuleSettingsField.objects.get(field_name='settings_field').value))
self.assertEquals('new_value', json.loads(XModuleSettingsField.objects.get(field_name='not_settings_field').value))
def test_delete_existing_field(self):
"Test that deleting an existing field removes it"
self.kvs.delete(settings_key('settings_field'))
self.assertEquals(0, XModuleSettingsField.objects.all().count())
def test_delete_missing_field(self):
"Test that deleting a missing field from an existing SettingsField raises a KeyError"
self.assertRaises(KeyError, self.kvs.delete, settings_key('not_settings_field'))
self.assertEquals(1, XModuleSettingsField.objects.all().count())
class TestContentStorage(TestCase):
def setUp(self):
content = ContentFactory.create()
self.user = UserFactory.create()
self.desc_md = {}
self.smc = StudentModuleCache(course_id, self.user, [])
self.kvs = LmsKeyValueStore(course_id, self.user, self.desc_md, self.smc)
def test_get_existing_field(self):
"Test that getting an existing field in an existing ContentField works"
self.assertEquals('content_value', self.kvs.get(content_key('content_field')))
def test_get_missing_field(self):
"Test that getting a missing field from an existing ContentField raises a KeyError"
self.assertRaises(KeyError, self.kvs.get, content_key('not_content_field'))
def test_set_existing_field(self):
"Test that setting an existing field changes the value"
self.kvs.set(content_key('content_field'), 'new_value')
self.assertEquals(1, XModuleContentField.objects.all().count())
self.assertEquals('new_value', json.loads(XModuleContentField.objects.all()[0].value))
def test_set_missing_field(self):
"Test that setting a new field changes the value"
self.kvs.set(content_key('not_content_field'), 'new_value')
self.assertEquals(2, XModuleContentField.objects.all().count())
self.assertEquals('content_value', json.loads(XModuleContentField.objects.get(field_name='content_field').value))
self.assertEquals('new_value', json.loads(XModuleContentField.objects.get(field_name='not_content_field').value))
def test_delete_existing_field(self):
"Test that deleting an existing field removes it"
self.kvs.delete(content_key('content_field'))
self.assertEquals(0, XModuleContentField.objects.all().count())
def test_delete_missing_field(self):
"Test that deleting a missing field from an existing ContentField raises a KeyError"
self.assertRaises(KeyError, self.kvs.delete, content_key('not_content_field'))
self.assertEquals(1, XModuleContentField.objects.all().count())
class TestStudentPrefsStorage(TestCase):
def setUp(self):
student_pref = StudentPrefsFactory.create()
self.user = student_pref.student
self.desc_md = {}
self.smc = StudentModuleCache(course_id, self.user, [])
self.kvs = LmsKeyValueStore(course_id, self.user, self.desc_md, self.smc)
def test_get_existing_field(self):
"Test that getting an existing field in an existing StudentPrefsField works"
self.assertEquals('student_pref_value', self.kvs.get(student_prefs_key('student_pref_field')))
def test_get_missing_field(self):
"Test that getting a missing field from an existing StudentPrefsField raises a KeyError"
self.assertRaises(KeyError, self.kvs.get, student_prefs_key('not_student_pref_field'))
def test_set_existing_field(self):
"Test that setting an existing field changes the value"
self.kvs.set(student_prefs_key('student_pref_field'), 'new_value')
self.assertEquals(1, XModuleStudentPrefsField.objects.all().count())
self.assertEquals('new_value', json.loads(XModuleStudentPrefsField.objects.all()[0].value))
def test_set_missing_field(self):
"Test that setting a new field changes the value"
self.kvs.set(student_prefs_key('not_student_pref_field'), 'new_value')
self.assertEquals(2, XModuleStudentPrefsField.objects.all().count())
self.assertEquals('student_pref_value', json.loads(XModuleStudentPrefsField.objects.get(field_name='student_pref_field').value))
self.assertEquals('new_value', json.loads(XModuleStudentPrefsField.objects.get(field_name='not_student_pref_field').value))
def test_delete_existing_field(self):
"Test that deleting an existing field removes it"
print list(XModuleStudentPrefsField.objects.all())
self.kvs.delete(student_prefs_key('student_pref_field'))
self.assertEquals(0, XModuleStudentPrefsField.objects.all().count())
def test_delete_missing_field(self):
"Test that deleting a missing field from an existing StudentPrefsField raises a KeyError"
self.assertRaises(KeyError, self.kvs.delete, student_prefs_key('not_student_pref_field'))
self.assertEquals(1, XModuleStudentPrefsField.objects.all().count())
class TestStudentInfoStorage(TestCase):
def setUp(self):
student_info = StudentInfoFactory.create()
self.user = student_info.student
self.desc_md = {}
self.smc = StudentModuleCache(course_id, self.user, [])
self.kvs = LmsKeyValueStore(course_id, self.user, self.desc_md, self.smc)
def test_get_existing_field(self):
"Test that getting an existing field in an existing StudentInfoField works"
self.assertEquals('student_info_value', self.kvs.get(student_info_key('student_info_field')))
def test_get_missing_field(self):
"Test that getting a missing field from an existing StudentInfoField raises a KeyError"
self.assertRaises(KeyError, self.kvs.get, student_info_key('not_student_info_field'))
def test_set_existing_field(self):
"Test that setting an existing field changes the value"
self.kvs.set(student_info_key('student_info_field'), 'new_value')
self.assertEquals(1, XModuleStudentInfoField.objects.all().count())
self.assertEquals('new_value', json.loads(XModuleStudentInfoField.objects.all()[0].value))
def test_set_missing_field(self):
"Test that setting a new field changes the value"
self.kvs.set(student_info_key('not_student_info_field'), 'new_value')
self.assertEquals(2, XModuleStudentInfoField.objects.all().count())
self.assertEquals('student_info_value', json.loads(XModuleStudentInfoField.objects.get(field_name='student_info_field').value))
self.assertEquals('new_value', json.loads(XModuleStudentInfoField.objects.get(field_name='not_student_info_field').value))
def test_delete_existing_field(self):
"Test that deleting an existing field removes it"
self.kvs.delete(student_info_key('student_info_field'))
self.assertEquals(0, XModuleStudentInfoField.objects.all().count())
def test_delete_missing_field(self):
"Test that deleting a missing field from an existing StudentInfoField raises a KeyError"
self.assertRaises(KeyError, self.kvs.delete, student_info_key('not_student_info_field'))
self.assertEquals(1, XModuleStudentInfoField.objects.all().count())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment